How to get rid of nested RxJava streams?

Question

I have a chain of calls to internet, database and as result I show collected info to user. Now I have very ugly 3-level nested RxJava stream. I really want to make it smooth and easy to read, but I've stuck really hard.

I already read everything about Map, flatMap, zip, etc. Cant' make things work together.

Code: make api call. Received info put in database subscribing to another stream in onSuccess method of first stream, and in onSuccess method of second stream received from DB info finally shows up to user.

Dat Frankenstein:

disposables.add(modelManager.apiCall()
                .subscribeOn(Schedulers.io())
                .observeOn(mainThread)
                .subscribeWith(new DisposableSingleObserver {

                  public void onSuccess(ApiResponse apiResponse) {

                        modelManager.storeInDatabase(apiResponse)
                       //level 1 nested stream:
                        disposables.add(modelManager.loadFromDatabas()
                                  .subscribeOn(Schedulers.io())
                                  .observeOn(mainThread)
                                  .subscribeWith(new DisposableSingleObserver{
                                    public void onSuccess(Data data) {
                                        view.showData(data);
                                    }
                                    public void onError(Throwable e) {
                                    }
                                }));
                    }
                    @Override
                    public void onError(Throwable e) {
                    }
                }));
    }

Terceira

Answer 1

I already read everything about Map, flatMap, zip, etc. Cant' make things work together.

Well, you missed something about flatMap, because this is what it's for ;)

---

disposables.add(
    modelManager.apiCall()
        .subscribeOn(Schedulers.io())
        .doOnSuccess((apiResponse) -> {
            modelManager.storeInDatabase(apiResponse)
        })
        .flatMap((apiResponse) -> {
            modelManager.loadFromDatabase()
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe((data) -> {
            view.showData(data);
        })
);

---

But if you use a reactive database layer like Room's LiveData<List<T>> support, then you can actually ditch the modelManager.loadFromDatabase() part.

Zebra

Actually, I am using Room database, but all queries to my DB return Single observables. I know one cool lib "room-with-a-view" and maybe I can find answer there. Thanks for direction! – Terceira

Wow! That sweet lambdas worked as charm! ) But I will dig further to setup LiveData and update UI according to DB changes. Thank for the answer! – Terceira

Answer 2

flatMap means convert the result of one stream into another stream, it is likely what you want.

Because you have an Observable that emits a ApiResponse then you have another "source of Observables" that takes this ApiResponse and gives another Observable that you want to observe on.

So you may likely want something like that:

disposables.add(modelManager.apiCall()
            .flatMap(apiResponse -> {
                         modelManager.storeInDatabase(apiResponse);
                         return modelManager.loadFromDatabas()
                     })
            .subscribeOn(Schedulers.io())
            .observeOn(mainThread)
            .subscribeWith(new DisposableSingleObserver {

              public void onSuccess(ApiResponse apiResponse) {
                  view.showData(data);
              }
              public void onError(Throwable e) {
              }
            })

Walkway