How to wait for Observable's onComplete() in flatMap()

Question

I have an Observable who reads data from a database. If data is null I need to get it from the network. So I do flatMap on first Observable, check the result of the database operation and if it is null I start that another Observable to fetch data from the network.

Note: Observables have different Subscribers because I have different postprocessing depending on where data comes from (such a logic).

 Observable.just(readDataFromDb()).flatMap(new Func1<SomeData, Observable<String>>() {
        @Override public Observable<SomeData> call(SomeData s) {
          if (s == null) {
            getReadFromNetworkObservable().subscribe(new AnotherSubscriber()); // this one might not complete
            return Observable.empty(); // I think I need to send this one only after readFromNetwork() completed
          } else {
            return Observable.just(s);
          }
        }
      }).subscribe(new SomeSubscirber());

Given I send Observable.empty() to exclude data processing for SomeSubscriber, I have a foreboding my second Observable can not always be finished because it might be simply garbage collected. I guess I saw it during my tests.

At this point, I think I just need to wait until Observable who reads from the network completed and then send Observable.empty(). So can I make the execution synchronous? But still I have a feeling I do it wrong.

Answer 1

You can make any observable as blocking with .toBlocking shortcut (See full info https://github.com/ReactiveX/RxJava/wiki/Blocking-Observable-Operators)

Data d = getReadFromNetworkObservable()
            .toBlocking()
            .first() //  or single() or singleOrDefault()

// manipulate with data here

Combining cache with network data is described here: http://blog.danlew.net/2015/06/22/loading-data-from-multiple-sources-with-rxjava/

And here: RxJava and Cached Data