如何使用 RxJava 从 Realm 获取带有对象列表的 Observable？答案

【问题标题】：How to get Observable with List of objects from Realm using RxJava?如何使用 RxJava 从 Realm 获取带有对象列表的 Observable？
【发布时间】：2017-10-09 22:21:27
【问题描述】：

我只有这个：

public Observable<List<Movie>> getAll() {
    return Observable.just(Movie.class)
            .flatMap(t -> Observable.just(t)
                    .doOnSubscribe(disposable -> realm.executeTransaction(realm1 -> realm1.where(Movie.class).findAll()))
                    .onErrorResumeNext((ObservableSource<? extends Class<Movie>>) observer -> Observable.empty())
                    .map(all -> realm.where(Movie.class).findAll())
            );
}

但它看起来真的很丑））

如果有可能避免重复代码并保存 realm.where(Movie.class).findAll() 并在 map() 中重用，一切都会好起来的。不推荐使用 RealmResults 方法 addAll。

【问题讨论】：

我认为您应该阅读this 并检查this 的答案。
我不认为这会起作用，因为 Observable.empty 只会发出 onComplete。所以你会得到一个完成的可观察对象。
@masp 感谢您链接我的答案，不幸的是，那里的提问者刚刚放弃了他的问题，所以我无法将其标记为重复。悲惨的生活……

标签： realm rx-java observable rx-java2

【解决方案1】：

也如here所述：

private io.reactivex.Flowable<List<Movie>> getAll() {
    return io.reactivex.Flowable.create(new FlowableOnSubscribe<List<Movie>>() {
        @Override
        public void subscribe(FlowableEmitter<List<Movie>> emitter)
                throws Exception {
            Realm realm = Realm.getDefaultInstance();
            RealmResults<Movie> results = realm.where(Movie.class).findAllAsync();
            final RealmChangeListener<RealmResults<Movie>> listener = _realm -> {
                if(!emitter.isUnsubscribed() && results.isLoaded()) {
                     emitter.onNext(results);
                }
            };
            emitter.setDisposable(Disposables.fromRunnable(() -> {
                results.removeChangeListener(listener);
                realm.close();
            }));
            results.addChangeListener(listener);
        }
    }, BackpressureStrategy.LATEST)
    .subscribeOn(AndroidSchedulers.mainThread())
    .unsubscribeOn(AndroidSchedulers.mainThread());

但正如@masp 所说，您可以在my article about this on realm.io that was published a month ago 中阅读更多关于使用Realm 和RxJava2 设计反应式数据层的信息。

因此，使用 Realm 4.0.0-RC1 及更高版本，您实际上可以这样做

private io.reactivex.Flowable<List<Movie>> getAll(Realm realm) {
    if(realm.isAutoRefresh()) {
        return realm.where(Movie.class)
                .findAllAsync()
                .asFlowable()
                .filter(RealmResults::isLoaded);
    } else { // for background threads
        return Flowable.just(realm.where(Movie.class).findAll());
    }
}

【讨论】：

【解决方案2】：

您可以随时将查找电影的代码提取到另一种方法中以重复使用。即使在函数式 API 中，我们也应该避免破坏 DRY。

public void code() {
    public Observable<List<Movie>> getAll () {
        return Observable.just(Movie.class)
                .flatMap(t -> Observable.just(t)
                        .doOnSubscribe(disposable -> realm.executeTransaction(realm1 -> findMovie((Object) realm1)))
                        .onErrorResumeNext((ObservableSource<? extends Class<Movie>>) observer -> Observable.empty())
                        .map(all -> findMovie((Object) all))
                );
    }

}

private Object findMovie(Object realm1) {
    return realm1.where(Movie.class).findAll();
}

【讨论】：