使用 rx-java amb() 或 switchIfEmpty() 时，领域 Observable 未完成

Question

当通过在查询上调用asObservable() 使用Realm Observable 时，使用amb() 或switchIfEmpty() 会导致领域的observable 无法完成其序列。可以通过使用 Observable.just() 而不是 Realms asObservable() 来解决此问题。

我不知道这是由我的代码还是 rx-java 或 Realm 中的错误引起的。

      mSubscription = getRealmObservable(params).switchIfEmpty(getNetworkObservable(params))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(subscriber);

}

private Observable<model> getNetworkObservable(UrlParams params) {
    final api service = NetworkManager.getAPI();
    return service.getModel(params.toMap())
            .doOnNext(new Action1<RealmList<Model>>() {
                @Override
                public void call(RealmList<Model> models) {
                    if (models != null && models.size() > 0) {
                        mRealm.beginTransaction();
                        mRealm.copyToRealmOrUpdate(models);
                        mRealm.commitTransaction();
                    }
                }
            })
            .flatMap(new Func1<RealmList<Model>, Observable<Model>>() {
                @Override
                public Observable<Model> call(RealmList<Model> models) {
                    return Observable.from(models);
                }
            });
}

private Observable<Model> getRealmObservable(final UrlParams params) {

    return Observable.just(mRealm.where(Model.class).findAll())/*.asObservable()*/ <- Using this cause this sequence not to finish
            .filter(new Func1<RealmResults<Model>, Boolean>() {
                @Override
                public Boolean call(RealmResults<Model> models) {
                    return models != null && models.isValid() && models.size() > 0;
                }
            })
            .flatMap(new Func1<RealmResults<Model>, Observable<Model>>() {
                @Override
                public Observable<Model> call(RealmResults<Model> models) {
                    return Observable.from(models);
                }
            });
}

Answer 1

在 RealmResults 上调用 asObservable() 创建的 observable 永远不会终止（调用 onComplete）。请参阅文档here。