chaining-rxjava-6-1.java

// NetworkRepository.java
@RxLogObservable
public Observable<Data> getData() {
    // implementation
}

// DiskRepository.java
@RxLogObservable
public Observable<Timestamped<Data>> getData() {
    // implementation
}
public void saveData(Timestamped<Data> data) {
    // implementation
}

// DomainService.java
@RxLogObservable
private Observable<Timestamped<Data>> getMergedData() {
  return Observable.mergeDelayError(
    diskRepository.getData().subscribeOn(Schedulers.io()),
    networkRepository.getData()
      .timestamp() // <-- This is where the magic happens
      .doOnNext(new Action1<Timestamped<Data>>() { 
        @Override 
        public void call(Timestamped<Data> data) { 
          diskRepository.saveData(data);
        } 
      }).subscribeOn(Schedulers.io())
  );
}

@RxLogObservable
public Observable<Timestamped<Data>> getData(IDisplayedData displayedData) {
  return getMergedData()
    .filter( 
      new Func1<Timestamped<Data>, Boolean>() { 
        @Override 
        public Boolean call(Timestamped<Data> timestampedData) { 
          return timestampedData != null &&
            timestampedData.getValue() != null &&
            // we check against the already displayed data's timestamp on each emission
            timestampedData.getTimestampMillis() > 
              displayedData.getTimestampMillis(); 
        }
      }
    );
}