Correct way to nest RxJava Observables?

Question

I'm using RxJava and nesting Observables as below. I want to call one observable inside another observable and have the outer observable emit the result of the inner observable onNext. It appears to work, but I'm not certain this is the correct implementation since I wasn't able to find any documentation to confirm.

public Observable<User> updateUser(final String id) {
    return Observable.create(new Observable.OnSubscribe<User>() {
        @Override
        public void call(final Subscriber<? super User> observer) {
                try {
                if (!observer.isUnsubscribed()) {
                    getUser(id).subscribe(new Action1<User>() {
                        @Override
                        public void call(User user) {
                            observer.onNext(user);
                                observer.onCompleted();
                            }
                        });
                    }
                } catch (Exception e) {
                    observer.onError(e);
                }
        }
    });
}

public Observable<User> getUser(final String id) {
...
}

I think the preferred method is to chain Observables, not nest them... — OneCricketeer
Why do you want to nest it, you could just return/call getUser directly? — akarnokd
I want to nest it because it makes the calling code simpler. The non-Observable version would be like public User updateUser(final String id) { //update user code return getUser(id); } — andrewe
But in your example both getUser and updateUser return Observable<User> and updateUser does nothing other than running getUser and emitting its value, so updateUser(id) { return getUser(id); } — akarnokd
Don't do that. It usually means unsubscription, resource cleanup and backpressure are broken and really hard to reason about. If you want to handle these things correctly, you codes will become much more complicated than chaining Observables/ — zsxwing

Dave Moten Dave Moten · Accepted Answer · 2016-01-13T10:50:38

Avoid calls to subscribe when you are just doing Observable transformations as you have to be mindful of all the problems mentioned by @akarnokd and @zsxwing in the comments.

I would also avoid using Observable.create because creating OnSubscribe implementations involves considering backpressure and consequent tricky business with concurrency. Prefer Observable.just,Observable.from,Observable.range,Observable.defer, Observable.using (there are more, check the wiki) and for more advanced purposes implement SyncOnSubscribe.

This code probably covers your use case:

public Observable<User> updateUser(final String id) {
    return getUser(id).doOnNext(user -> updateUser(user));
}

public void updateUser(User user) {
    //whatever you want here
}

public Observable<User> getUser(final String id) {
    ...
}

Correct way to nest RxJava Observables?

1 Answers