Chaining-rxjava-4-2.java

// DomainService.java
@RxLogObservable
public Observable<Data> getMergedData() {
  return Observable.mergeDelayError( // <-- don't interrupt stream
    diskRepository.getData().subscribeOn(Schedulers.io()),
    networkRepository.getData()
      .doOnNext(new Action1<Data>() { 
        @Override 
        public void call(Data data) { 
          diskRepository.saveData(data);
        } 
      }).subscribeOn(Schedulers.io())
  );
}