title: Andorid rxJava: how to get data from cache and and the same time update it in the background?
body: <p>I just start learning <strong>rxJava</strong> for Android and want to implement the common use case:</p>
  
  <ul>
  <li>request data from cache and show to the user </li>
  <li>request data from web</li>
  <li>server update data in storage and automatically show it to the user</li>
  </ul>
  
  <p>Traditionally on of the best scenarios was use <strong>CursorLoader</strong> to get data from cache, run web request in the separate thread and save data to the disk via <strong>content provider</strong>, <strong>content provider</strong> automatically notify the listener and <strong>CursorLoader</strong> autoupdate UI. </p>
  
  <p>In rxJava I can do it by running two different Observers as you can see in code below, but I don't find the way how to combine this two calls into the one to reach my aim. Googling shows this thread but it looks like it just get data from the cache or data from the web server, but don't do both <a href="https://stackoverflow.com/questions/26939175/rxjava-and-cached-data">RxJava and Cached Data</a></p>
  
  <p>Code snippet: </p>
  
  <pre><code>@Override
  public Observable&lt;SavingsGoals&gt; getCachedSavingsGoal() {
      return observableGoal.getSavingsGoals()
              .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread());
  }
  
  @Override
  public Observable&lt;SavingsGoals&gt; getRecentSavingsGoal() {
      return api.getSavingsGoals()
              .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread());
  }
  
      model.getCachedSavingsGoal().subscribe(new Observer&lt;SavingsGoals&gt;() {
          @Override
          public void onCompleted() {
              // no op
          }
  
          @Override
          public void onError(Throwable e) {
              Log.e(App.TAG, "Failed to consume cached data");
              view.showError();
          }
  
          @Override
          public void onNext(SavingsGoals savingsGoals) {
              Log.d(App.TAG, "Show the next item");
              if (savingsGoals != null &amp;&amp; !savingsGoals.getSavingsGoals().isEmpty()) {
                  view.showData(savingsGoals.getSavingsGoals());
              } else {
                  view.showError();
              }
          }
      });
  
      model.getRecentSavingsGoal().subscribe(new Observer&lt;SavingsGoals&gt;() {
          @Override
          public void onCompleted() {
              // no op
          }
  
          @Override
          public void onError(Throwable e) {
              Log.e(App.TAG, "Failed to consume data from the web", e);
              view.showError();
          }
  
          @Override
          public void onNext(SavingsGoals savingsGoals) {
              if (savingsGoals != null &amp;&amp; !savingsGoals.getSavingsGoals().isEmpty()) {
                  view.showData(savingsGoals.getSavingsGoals());
              } else {
                  view.showError();
              }
          }
      });
  </code></pre>
  
  <p>Also, the one of issues with current approach is cache and web data are not garranted to be run sequently. It is possible when outdated data will come as latest and override recent from web.</p>
  
  <p>To solve this issue I implemented Observer merge with filtration by timestamp: it get data from cache, pass it to the next observer and if cache is outdated fire new call to the web - case for thread competition solved by the filtration with timestamps. However, the issue with this approach I can not return cache data from this Observable - I need to wait when both requests finish their work. </p>
  
  <p>Code snippet. </p>
  
  <pre><code>    @Override
  public Observable&lt;Timestamped&lt;SavingsGoals&gt;&gt; getSavingGoals() {
      return observableGoal
              .getTimestampedSavingsGoals()
              .subscribeOn(Schedulers.io())
              .flatMap(new Func1&lt;Timestamped&lt;SavingsGoals&gt;, Observable&lt;Timestamped&lt;SavingsGoals&gt;&gt;&gt;() {
                  @Override
                  public Observable&lt;Timestamped&lt;SavingsGoals&gt;&gt; call(Timestamped&lt;SavingsGoals&gt; cachedData) {
                      Log.d(App.FLOW, "getTimestampedSavingsGoals");
                      return getGoalsFromBothSources()
                              .filter(filterResponse(cachedData));
                  }
              })
              .subscribeOn(AndroidSchedulers.mainThread());
  }
  
  private Func1&lt;Timestamped&lt;SavingsGoals&gt;, Boolean&gt; filterResponse(Timestamped&lt;SavingsGoals&gt; cachedData) {
      return new Func1&lt;Timestamped&lt;SavingsGoals&gt;, Boolean&gt;() {
          @Override
          public Boolean call(Timestamped&lt;SavingsGoals&gt; savingsGoals) {
              return savingsGoals != null
                      &amp;&amp; cachedData != null
                      &amp;&amp; cachedData.getTimestampMillis() &lt; savingsGoals.getTimestampMillis()
                      &amp;&amp; savingsGoals.getValue().getSavingsGoals().size() != 0;
          }
      };
  }
  
  private Observable&lt;Timestamped&lt;SavingsGoals&gt;&gt; getGoalsFromBothSources() {
      Log.d(App.FLOW, "getGoalsFromBothSources:explicit");
      return Observable.merge(
              observableGoal.getTimestampedSavingsGoals().subscribeOn(Schedulers.io()),
              api.getSavingsGoals()
                      .timestamp()
                      .flatMap(new Func1&lt;Timestamped&lt;SavingsGoals&gt;, Observable&lt;Timestamped&lt;SavingsGoals&gt;&gt;&gt;() {
                          @Override
                          public Observable&lt;Timestamped&lt;SavingsGoals&gt;&gt; call(Timestamped&lt;SavingsGoals&gt; savingsGoals) {
                              Log.d(App.FLOW, "getGoalsFromBothSources:implicit");
                              return observableGoal.saveAllWithTimestamp(savingsGoals.getTimestampMillis(), savingsGoals.getValue().getSavingsGoals());
                          }
                      }))
                      .subscribeOn(Schedulers.io());
  }
  </code></pre>
  
  <p>Do you know the approach to do this in one Observer? </p>
  
  <p><strong>Potential solution:</strong></p>
  
  <pre><code>@Override
  public Observable&lt;SavingsGoals&gt; getSavingGoals() {
      return api.getSavingsGoals()
              .publish(network -&gt;
                      Observable.mergeDelayError(
                              observableGoal.getSavingsGoals().takeUntil(network),
                              network.flatMap(new Func1&lt;SavingsGoals, Observable&lt;SavingsGoals&gt;&gt;() {
                                  @Override
                                  public Observable&lt;SavingsGoals&gt; call(SavingsGoals savingsGoals) {
                                      return observableGoal.saveAll(savingsGoals.getSavingsGoals());
                                  }
                              })
                      )
              )
              .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread());
  }
  </code></pre>
  
  <ul>
  <li>Sorry, hot replacement in IDE hide the issue which this approach has: first one in case if network unavailable and cache thread completes first, the error will terminate whole merge (solved by mergeDelayError), second one is in case when cache is empty and returns first data from web request will not be return on subscriber. As you can see my method returns Observable after save and traditional merge as I shown in my code properly handle this case but takeUntil by some reason can not. Question is still open. </li>
  </ul>