murki

// wrapping synchronous operation in an RxJava Observable
Observable<Boolean> wipeContents(final SharedPreferences sharedPreferences) { 
    return Observable.fromCallable(new Callable<Boolean>() { 
        @Override 
        public Boolean call() throws Exception { 
            return sharedPreferences.edit().clear().commit(); 
        } 
    }); 
}

murki

// DomainService.java
@RxLogObservable
public Observable<Data> getData() {
  return getMergedData()
    .onErrorReturn(new Func1<Throwable, Data>() {
      @Override
      public Data call(Throwable throwable) {
        Log.e(CLASSNAME, "Error while fetching data. Swallowing the exception.", throwable);
        return null; // We return null since we know our filter will ignore null values.
      }

murki

// We assume the repositories getData() methods will return null if no data is available

// DomainService.java
@RxLogObservable
public Observable<Data> getData() {
  return getMergedData()
    .filter( 
      new Func1<Data, Boolean>() { 
        @Override 
        public Boolean call(Data data) { 

murki

// DiskRepository.java
public void saveData(Data data) {
    // implementation
}

// DomainService.java
@RxLogObservable
public Observable<Data> getMergedData() {
  return Observable.merge(
    diskRepository.getData().subscribeOn(Schedulers.io()),

murki

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

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

murki

// This assumes both 'repositories' return Observable<Data>

Observable<Data> results = Observable.merge(
  diskRepository.getData().subscribeOn(Schedulers.io()),
  networkRepository.getData().subscribeOn(Schedulers.io())
)

murki

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

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