Realm, RxJava, asObservable() and doOnUnsubscribe()

Question

In my Android projects, I use realm as my data storage engine. I love it!
I also use RxJava because it makes "threading" so much easier, and I really like the whole "reactive mindset". I love it!

I use an MVP pattern + some "Clean architecture" ideas to build my apps.

My Interactors are the only ones who know about Realm. I expose data with the help of Observable, like this:

@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>());
}

The problem is my doOnUnsubscribe side-effect gets called before Realm can do its thing, handling the exposed 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)

I created a sandbox project for this use case.

I really like using Realm+RxJava, but I can't seem to find a clean solution to close the Realm instance when I unsubscribe (I usually unsubscribe when the activity gets destroyed). Any ideas?

Edit 1: https://github.com/realm/realm-java/issues/2357
Edit 2: thanks to the very active realm team, there is already a pull request to fix this issue.

Answer 1

21 hours later and this is what I came up with:

@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);
        }
    });
}

I tried something like this before posting the question on stackoverflow, but my mistake was using flatMap(), instead of concatMap().

Unlike flatMap(),concatMap() will keep the order of the emissions which, in my case, means that my Action0 -> realm.close() will be the last action being called after unsubscribing from the stream, after Realm's Action0 -> results.removeChangeListener(listener) which was causing the problem.

A full example can be found on github.

Edit: thanks to the very active realm team, there is already a pull request to fix this issue.