如何将 Realm asObservable 与 RxJava 的 concat() 运算符一起使用？答案

【问题标题】：How to use Realm asObservable with RxJava's concat() operator?如何将 Realm asObservable 与 RxJava 的 concat() 运算符一起使用？
【发布时间】：2015-12-23 14:54:54
【问题描述】：

我正在尝试将 Realm 与 RxJava 和 Retrofit 一起使用，DanLew 描述了 here 连接来自领域和改造的输入，但如果我将领域添加到链中，它就会卡住

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

存储

 @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());
      }

似乎这种情况可能会发生，因为可观察对象是 Realm 的热门对象，但文档中没有关于它的内容以及我认为如何将 Realm 与其他可观察对象组合？

更新： 事实证明它以旧方式工作正常。关于新 api 的问题仍然存在。

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

【问题讨论】：

github.com/realm/realm-java/issues/1998 github 上的相同问题。在那里查看进展。
我也有同样的问题。您找到解决方案了吗？
doOnCompleted 永远不会被调用
您确定，您的 Realm 数据库中有一些国家/地区吗？如果不是，则链仍停留在.first(countries -> return !countries.isEmpty()) 行

标签： android realm rx-java

【解决方案1】：

这是因为countryStorage.restoreAsObservable() 永远不会完成，如果您阅读concat 文档，它明确指出：

Concat 等待订阅您传递给它的每个其他 Observable，直到前一个 Observable 完成。

相反，您可以执行以下操作：

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

【讨论】：