使用Realm和RxJava 2 [英] Using Realm with RxJava 2

问题描述

我在Android应用程序中使用RxJava 2，并且正在集成Realm。据我所知，Realm默认只支持RxJava 1，并且在查询 RealmResults<？> <时允许返回 Observable / code>，如下所示：

I am using RxJava 2 in my Android application, and am integrating Realm. As far as I can tell, Realm only supports RxJava 1 by default, and allows an Observable to be returned when querying for RealmResults<?>, like so:

Realm.getDefaultInstance()
    .where(VideoBundle.class)
    .findAll()
    .asObservable()
    .first()

返回的Observable来自RxJava 1.我如何一起使用Realm和RxJava 2？我遇到了两个相关问题，发现这里和这里，但没有找到简洁的答案。此外，文档（在此处找到： https://realm.io/docs/java/latest /＃rxjava ）提到创建自定义 RxObservableFactory ，但没有提供有关如何执行此操作的资源。

The Observable returned is from RxJava 1. How can I use Realm and RxJava 2 together? I have come across 2 relevant issues, found here and here, but no succinct answer was found. Additionally, the documentation (found here: https://realm.io/docs/java/latest/#rxjava) mentions creating a custom RxObservableFactory, but provides no resources on how to do so.

Realm如何与已经使用RxJava 2的项目一起使用？

How can Realm be used with a project already using RxJava 2?

推荐答案

解决方案是使用Flowable包装RealmResults，最新的背压策略。

The solution is to wrap RealmResults with Flowable, with LATEST backpressure strategy.

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(...);

这篇关于使用Realm和RxJava 2的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！