Chaining RxJava observables with callbacks/listeners

Question

I am using Retrofit with Observables, and would like to chain the observables. Usually it works well with functions like map() or flatMap(), since the api returns an Observable that does the task. But in this case I have to do the following:

getKey() from the api
Use the returned key in another library Foo and wait for the callback to be called.
When the callback returns, send the result to the api.

I'd like this to be a single chained call so that I just have to subscribe once. I'm guessing I can use merge() or join() or something, but wasn't sure what the best approach would be to handle the callback.

Is there a way to make this better? This is what I have so far:

api.getKey().subscribe(new Action1<String>() {
   @Override
   public void call(String key) {
      Foo foo = new Foo();
      foo.setAwesomeCallback(new AwesomeCallback() {
         @Override
         public void onAwesomeReady(String awesome) {
            api.sendAwesome(awesome)
                    .subscribe(new Action1<Void>() {
                       @Override
                       public void call(Void aVoid) {
                           handleAwesomeSent();
                       }
                    });
         }
      });
      foo.makeAwesome();
   }
});

Where is the key used?

ersin-ertan
– ersin-ertan

2016-08-26 00:18:39 +00:00
Commented Aug 26, 2016 at 0:18 — ersin-ertan
– ersin-ertan, Commented Aug 26, 2016 at 0:18

alexwen · Accepted Answer · 2015-04-16 19:24:31Z

18

Adapting clemp6r's solution, here is another one that needs neither Subjects nor nested Subscriptions:

api.getKey().flatMap(new Func1<String, Observable<String>>() {
    @Override
    public Observable<String> call(String key) {

        return Observable.create(new Observable.OnSubscribe<String>(){

            @Override
            public void call(final Subscriber<? super String> subscriber) {
                Foo foo = new Foo();
                foo.setAwesomeCallback(new AwesomeCallback() {
                    @Override
                    public void onAwesomeReady(String awesome) {
                        if (! subscriber.isUnsubscribed()) {
                            subscriber.onNext(awesome);
                            subscriber.onComplete();
                        }
                    }
                });
                foo.makeAwesome();
            } 
        });
}).flatMap(new Func1<String, Observable<String>>() {
    @Override
    public Observable<String> call(String awesome) {
        return sendAwesome(awesome);
   }
}).subscribe(new Action1<Void>() {
    @Override
    public void call(Void aVoid) {
        handleAwesomeSent();
    }
});

In general I think it always possibly to wrap any callback based asynchronous operation in an Observable using Observable.create().

edited Apr 16, 2015 at 19:24

alexwen

1,1287 silver badges16 bronze badges

answered Apr 16, 2015 at 18:11

david.mihola

13.1k11 gold badges51 silver badges76 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Revan Over a year ago

This is very helpful

clemp6r · Accepted Answer · 2015-04-16 17:02:16Z

4

You have to use a PublishSubject to transform the callback-based API to an observable.

Try something like that (not tested):

api.getKey().flatMap(new Func1<String, Observable<String>>() {
   @Override
   public Observable<String> call(String key) {
      Foo foo = new Foo();
      PublishSubject<String> subject = PublishSubject.create();
      foo.setAwesomeCallback(new AwesomeCallback() {
         @Override
         public void onAwesomeReady(String awesome) {
            subject.onNext(awesome);
            subject.onComplete();
         }
      });
      foo.makeAwesome();

      return subject;
   }
}).flatMap(new Func1<String, Observable<String>>() {
   @Override
   public Observable<String> call(String awesome) {
       return sendAwesome(awesome);
   }
}).subscribe(new Action1<Void>() {
    @Override
    public void call(Void aVoid) {
        handleAwesomeSent();
    }
});

edited Apr 16, 2015 at 17:02

answered Apr 16, 2015 at 15:48

clemp6r

3,7532 gold badges28 silver badges34 bronze badges

2 Comments

alexwen Over a year ago

You will want to call subject.onComplete() after your subject.onNext(awesome) so the Observable chain will complete.

Artem Zinnatullin Over a year ago

PublishSubject is not required, you can use flatMap + Observable.create().

ersin-ertan · Accepted Answer · 2016-08-26 01:57:47Z

Api api = new Api() {
  @Override Single<String> getKey() {
    return Single.just("apiKey");
  }
};

api.getKey()
    .flatMap(key -> Single.<String>create( singleSubscriber -> {
        Foo foo = new Foo();
        foo.setAwesomeCallback(awesome -> {
          try { singleSubscriber.onSuccess(awesome);}
          catch (Exception e) { singleSubscriber.onError(e); }
        });
        foo.makeAwesome();
    }))
    .flatMapCompletable(
        awesome -> Completable.create(completableSubscriber -> {
          try {
            sendAwesome(awesome);
            completableSubscriber.onCompleted();
          } catch (Exception e) { completableSubscriber.onError(e); }
        }))
    .subscribe(this::handleAwesomeSent, throwable -> { });

See gist for full anonymous class example

This implementation adapts david.mihola answer by making use of Single and Completable types along with the flatMapCompletable(), while being type safe/specific.

Collectives™ on Stack Overflow

Chaining RxJava observables with callbacks/listeners

3 Answers 3

1 Comment

2 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

2 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related