我在 Android 应用程序中使用 RxJava 2,并且正在集成 Realm。据我所知,Realm 默认情况下仅支持 RxJava 1,并允许 Observable
查询 RealmResults<?>
时返回,像这样:
Realm.getDefaultInstance()
.where(VideoBundle.class)
.findAll()
.asObservable()
.first()
返回的Observable来自RxJava 1。如何一起使用Realm和RxJava 2?我遇到了2个相关问题,发现here和 here ,但没有找到简洁的答案。此外,文档(此处找到: https://realm.io/docs/java/latest/#rxjava )提到创建自定义 RxObservableFactory
,但没有提供有关如何执行此操作的资源。
如何将 Realm 与已使用 RxJava 2 的项目一起使用?
最佳答案
解决方案是使用 Flowable 包装 RealmResults,并使用最新的背压策略。
private io.reactivex.Flowable<RealmResults<_>> getSomeItems() {
return io.reactivex.Flowable.create(new FlowableOnSubscribe<RealmResults<__>>() {
@Override
public void subscribe(FlowableEmitter<RealmResults<__>> emitter)
throws Exception {
Realm realm = Realm.getDefaultInstance();
RealmResults<__> results = realm.where(__.class).findAllSortedAsync("__");
final RealmChangeListener<RealmResults<__>> 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());
从 Realm 4.0.0-RC1 及更高版本开始,我上面展示的这种行为是使用 realmResults.asFlowable()
实现的。
Disposable subscription = realm.where(__.class)
.findAllSortedAsync("__")
.asFlowable()
.filter(RealmResults::isLoaded)
.subscribe(...);
关于java - 将 Realm 与 RxJava 2 结合使用,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/44338239/