Realm access from incorrect thread on compositeDisposable clear

Question

I am adding all my observer that are subscribed in an activity are added to CompositeDisposable. OnStop of activity is calling mCompositeDisposable.clear() which creates below crash log. As CompositeDisposable.clear calls onDispose please find below code does to realm

Single.create(...).subscribeOn(Schedulers.io())
                   .observeOn(AndroidSchedulers.mainThread())
                   .subscribe(observer);
mCompositeDisposable.add(observer);

creats realm here in Schedulers.io()

Single<RealmList<T>> source ...
source .doOnDispose(() -> {
 if (mRealm == null) {
            return;
        }
 if (Looper.myLooper() != null) {
            mRealm.removeAllChangeListeners();
        }
 if (!mRealm.isClosed()) {
            mRealm.close();
        }
mRealm = null;
}  

Giving the below crash logs

java.lang.IllegalStateException: Realm access from incorrect thread.

Realm access from incorrect thread.
 Realm objects can only be accessed on the thread they were created.
                                                       at io.realm.BaseRealm.JN(SourceFile:438)
                                                       at io.realm.BaseRealm.removeAllListeners(SourceFile:263)
                                                       at io.realm.Realm.removeAllChangeListeners(SourceFile:1399)
...

 at io.reactivex.internal.operators.single.SingleDoOnDispose$DoOnDisposeObserver.dispose(SourceFile:60)
                                                       at io.reactivex.internal.operators.single.SingleDoFinally$DoFinallyObserver.dispose(SourceFile:85)
                                                       at io.reactivex.internal.disposables.DisposableHelper.aq(SourceFile:124)
                                                       at io.reactivex.internal.operators.single.SingleSubscribeOn$SubscribeOnObserver.dispose(SourceFile:78)
                                                       at io.reactivex.internal.disposables.DisposableHelper.aq(SourceFile:124)
                                                       at io.reactivex.internal.operators.single.SingleSubscribeOn$SubscribeOnObserver.dispose(SourceFile:78)
                                                       at io.reactivex.internal.disposables.DisposableHelper.aq(SourceFile:124)
                                                       at io.reactivex.internal.operators.single.SingleObserveOn$ObserveOnSingleObserver.dispose(SourceFile:87)
                                                       at io.reactivex.internal.disposables.DisposableHelper.aq(SourceFile:124)
                                                       at io.reactivex.observers.DisposableObserver.dispose(SourceFile:91)
                                                       at io.reactivex.disposables.CompositeDisposable.a(SourceFile:240)
                                                       at io.reactivex.disposables.CompositeDisposable.clear(SourceFile:206)

Tried with .unsubscribeOn(Schedulers.io()) but didnt worked

Answer 1

creats realm here in Schedulers.io()

You're creating instance of Realm in mainThread then asking Realm to operate (e.g. load data) in io thread which is impossible in Realm by design. Example by developers of Realm states few points to keep in mind here https://github.com/realm/realm-java/tree/master/examples/rxJavaExample.

Implementing 1st and 4th points into your code snippet we will get us proper solution. Note that method findAllAsync() will do everything asynchronously for you no need to create io thread by using RxJava

    Single<RealmList<T>> realmResults = realm.where(....)
                .equalTo(...)
                .findAllAsync()
                .asFlowable()
                .map { realmResult->
                    val realmList = RealmList<T>()
                    realmList.addAll(realmResult.toList())
                    return realmList
                }
                .firstOrError()

            realm.close()