Come utilizzare Realm asObservable con l'operatore concat() di RxJava?

Question

sto cercando di utilizzare Realm con RxJava e Retrofit in modo DanLew descritto here concating ingresso dal regno e retrofit, ma si blocca se l'aggiunta di regno nella catena

Observable.concat(countryStorage.restoreAsObservable(), 
       networkService.api() 
        .getCountries() 
        .doOnNext(countryStorage::save)) 
       .first() 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(//never reaching here) 

stoccaggio

@Override public Observable<List<Country>> restoreAsObservable() { 
     Realm realm = realmProvider.get(); 
     return realm.where(Country.class) 
      .findAll() 
      .asObservable() 
      .map(countries -> return realm.copyFromRealm(countries)) 
      .first(countries -> return !countries.isEmpty()) 
      .doOnCompleted(realm::close()); 
     } 

Sembra che questo potrebbe accadere che osservabile sia caldo dal Reame, ma nulla nei documenti e come suppongo di comporre il Reame con altri osservabili?

UPDATE: Si scopre che funziona perfettamente nel vecchio modo. La domanda rimane ancora sulle nuove API.

return Observable.just(
     realm.copyFromRealm(realm.where(Country.class).findAll())) 
     .filter(countries -> !countries.isEmpty()) 
     .doOnCompleted(realm::close); 

Answer 1

Sta accadendo perché countryStorage.restoreAsObservable() mai completa e se leggete concat doc, si afferma esplicitamente che:

Concat attende di abbonarsi a ogni ulteriore osservabile che si passa ad esso fino a quando le precedenti completa l'osservabili .

Invece si può semplicemente fare qualcosa di simile:

countryStorage.restoreAsObservable() 
      .doOnSubscribe(() -> { 
       networkService.api() 
        .getCountries() 
        .subscribe(countryStorage::save) 
      }) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(//do smth)