title: Combine database and network call with RxJava2
body: <p>I have 2 data sources: database (cache) and api and I need to combine them into one stream. I know that I can simply use concatArray or something similar but I want to achieve more complicated behavior:</p>
  
  <ul>
  <li><p>Observable stream which will emit up to 2 elements.</p></li>
  <li><p>It will subscribe to both sources in the beginning.</p></li>
  <li><p>If api call will be fast enough (&lt;~300ms), it will emit only data from it and complete the stream.</p></li>
  <li>If api call will be slow (>~300ms), emit data from database and still wait for data from api</li>
  <li>If api call won't succeed, emit data from database and emit an error.</li>
  <li>If database somehow will be slower than api, it can't emit its data (stream completion solves it)</li>
  </ul>
  
  <p>I accomplished it with the following code:</p>
  
  <pre><code>    public Observable&lt;Entity&gt; getEntity() {
      final CompositeDisposable disposables = new CompositeDisposable();
      return Observable.&lt;Entity&gt;create(emitter -&gt; {
          final Entity[] localEntity = new Entity[1];
  
          //database call:
          disposables.add(database.getEntity()
                  .subscribeOn(schedulers.io())
                  .doOnSuccess(entity -&gt; localEntity[0] = entity) //saving our entity because 
                                                          //apiService can emit error before 300 ms 
                  .delay(300, MILLISECONDS)
                  .subscribe((entity, throwable) -&gt; {
                      if (entity != null &amp;&amp; !emitter.isDisposed()) {
                          emitter.onNext(entity);
                      }
                  }));
  
          //network call:
          disposables.add(apiService.getEntity()
                  .subscribeOn(schedulers.io())
                  .onErrorResumeNext(throwable -&gt; {
                      return Single.&lt;Entity&gt;error(throwable) //we will delay error here
                              .doOnError(throwable1 -&gt; {
                                  if (localEntity[0] != null) emitter.onNext(localEntity[0]); //api error, emit localEntity
                              })
                              .delay(200, MILLISECONDS, true); //to let it emit localEntity before emitting error
                  })
                  .subscribe(entity -&gt; {
                      emitter.onNext(entity); 
                      emitter.onComplete(); //we got entity from api, so we can complete the stream
                  }, emitter::onError));
      })
              .doOnDispose(disposables::clear)
              .subscribeOn(schedulers.io());
  }
  </code></pre>
  
  <p>Code is a bit clunky and I'm creating here observables inside observable, which I think is <strong>bad</strong>. But that way I have global access to emitter, which allows me to control main stream (emit data, success, error) in the way I want.</p>
  
  <p>Is there better way to achieve this? I'd love to see some code examples. Thanks!</p>