android - Realm 、RxJava、asObservable() 和 doOnUnsubscribe()

Question

在我的 Android 项目中，我使用 realm作为我的数据存储引擎。我喜欢它!
我也用 RxJava因为它使“线程化”变得容易得多，而且我真的很喜欢整个“响应式(Reactive)思维方式”。我喜欢它!

我使用 MVP 模式 + 一些“清洁架构”的想法来构建我的应用。

我的 Interactors 是唯一了解 Realm 的人。我借助 Observable 公开数据，如下所示:

@Override
public Observable<City> getHomeTown() {
    final Realm realm = Realm.getDefaultInstance();
    return realm.where(City.class).equalTo("name", "Cluj-Napoca").findAllAsync().asObservable()
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    realm.close();
                }
            })
            .compose(new NullIfNoRealmObject<City>());
}

问题是我的 doOnUnsubscribe 副作用在 Realm 可以做它的事情之前被调用，处理暴露的 observable:

Caused by: java.lang.IllegalStateException: This Realm instance has already been closed, making it unusable.
at io.realm.BaseRealm.checkIfValid(BaseRealm.java:344)
at io.realm.RealmResults.removeChangeListener(RealmResults.java:818)
at io.realm.rx.RealmObservableFactory$3$2.call(RealmObservableFactory.java:137)
at rx.subscriptions.BooleanSubscription.unsubscribe(BooleanSubscription.java:71)
at rx.internal.util.SubscriptionList.unsubscribeFromAll(SubscriptionList.java:124)
at rx.internal.util.SubscriptionList.unsubscribe(SubscriptionList.java:113)
at rx.Subscriber.unsubscribe(Subscriber.java:98)
at rx.internal.util.SubscriptionList.unsubscribeFromAll(SubscriptionList.java:124)
at rx.internal.util.SubscriptionList.unsubscribe(SubscriptionList.java:113)
at rx.Subscriber.unsubscribe(Subscriber.java:98)
at rx.subscriptions.CompositeSubscription.unsubscribeFromAll(CompositeSubscription.java:150)
at rx.subscriptions.CompositeSubscription.unsubscribe(CompositeSubscription.java:139)
at ro.tudorluca.realm.sandbox.city.CityPresenter.onDestroy(CityPresenter.java:62)
at ro.tudorluca.realm.sandbox.city.CityActivity.onDestroy(CityActivity.java:35)

我创建了一个 sandbox此用例的项目。

我真的很喜欢使用 Realm+RxJava，但是我似乎找不到一个干净的解决方案来在我取消订阅 时关闭 Realm 实例(我通常在 Activity 被破坏)。有什么想法吗？

编辑 1:https://github.com/realm/realm-java/issues/2357
编辑 2:感谢非常活跃的 Realm 团队，已经有一个 pull request解决这个问题。

Answer 1

21 小时后，这就是我想出的:

@Override
public Observable<City> getHomeTown() {
    return getManagedRealm()
            .concatMap(new Func1<Realm, Observable<City>>() {
                @Override
                public Observable<City> call(Realm realm) {
                    return realm.where(City.class).equalTo("name", "Cluj-Napoca").findAllAsync().asObservable()
                            .compose(new NullIfNoRealmObject<City>());
                }
            });
}

private static Observable<Realm> getManagedRealm() {
    return Observable.create(new Observable.OnSubscribe<Realm>() {
        @Override
        public void call(final Subscriber<? super Realm> subscriber) {
            final Realm realm = Realm.getDefaultInstance();
            subscriber.add(Subscriptions.create(new Action0() {
                @Override
                public void call() {
                    realm.close();
                }
            }));
            subscriber.onNext(realm);
        }
    });
}

在将问题发布到 stackoverflow 之前，我尝试过类似的操作，但我的错误是使用了 flatMap()，而不是 concatMap()。

与 flatMap() 不同，concatMap() 将保持排放的顺序，在我的例子中，这意味着我的 Action0 -> realm.close () 将是取消订阅流后调用的最后一个操作，在导致问题的 Realm 的 Action0 -> results.removeChangeListener(listener) 之后。

可以在 github 上找到完整的示例.

编辑:感谢非常活跃的 Realm 团队，已经有一个pull request解决这个问题。