Rxjava + Realm access from incorrect thread

Question

I'm getting this exception reading/writing from Realm

06-19 09:49:26.352 11404-11404/****** E/ContentValues: loadData: OnError Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created. java.lang.IllegalStateException: Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created. at io.realm.BaseRealm.checkIfValid(BaseRealm.java:385) at io.realm.RealmResults.isLoaded(RealmResults.java:115) at io.realm.OrderedRealmCollectionImpl.size(OrderedRealmCollectionImpl.java:307) at io.realm.RealmResults.size(RealmResults.java:60) at java.util.AbstractCollection.isEmpty(AbstractCollection.java:86) at /****** .lambda$loadData$0(SplashPresenter.java:42) at /****** $$Lambda$1.test(Unknown Source) at io.reactivex.internal.operators.observable.ObservableFilter$FilterObserver.onNext(ObservableFilter.java:45) at io.reactivex.observers.SerializedObserver.onNext(SerializedObserver.java:111) at io.reactivex.internal.operators.observable.ObservableDelay$DelayObserver$1.run(ObservableDelay.java:84) at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:59) at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:51) at java.util.concurrent.FutureTask.run(FutureTask.java:237) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607) at java.lang.Thread.run(Thread.java:761)

This is the code:

  mSubscribe = Observable.just(readData())
            .delay(DELAY, TimeUnit.SECONDS)
            .filter(value -> !value.isEmpty())
            .switchIfEmpty(createRequest())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread()).subscribe(data -> {
                getView().hideLoading();
                writeData(data);
            }, 
           (throwable -> {
            }));

Read data

  private List<CategoryModel> readData() {
    Realm defaultInstance = Realm.getDefaultInstance();
    List<CategoryModel> title = defaultInstance.where(CategoryModel.class).findAllSorted("title");

    defaultInstance.close();
    return title;
}

Write data

private void writeData(List<CategoryModel> categoryModels) {

        try {
            Realm defaultInstance = Realm.getDefaultInstance();
            defaultInstance.executeTransactionAsync(realm -> realm.insertOrUpdate(categoryModels));
            defaultInstance.close();
        } finally {
            getView().notifyActivity(categoryModels);
        }
    }

How can I follow this logic using the proper threads?

Answer 1

The only rule to using Realm across threads is to remember that Realm, RealmObject or RealmResults instances cannot be passed across threads.

When you want to access the same data from a different thread, you should simply obtain a new Realm instance (i.e. Realm.getDefaultInstance()) and get your objects through a query (then close Realm at the end of the thread).

The objects will map to the same data on disk, and will be readable & writeable from any thread! You can also run your code on a background thread using realm.executeTransactionAsync() like this .

Answer 2

How can i follow this logic using the proper threads?

By not trying to read on Schedulers.io() for your UI thread (Realm gives auto-updating lazy-loaded proxy views that provide change notifications for your data on the UI thread, after all).

---

So instead of this

 mSubscribe = Observable.just(readData())
        .delay(DELAY, TimeUnit.SECONDS)
        .filter(value -> !value.isEmpty())
        .switchIfEmpty(createRequest())
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread()).subscribe(data -> {
            getView().hideLoading();
            writeData(data);
        }, 
       (throwable -> {
        }));

private List<CategoryModel> readData() {
    Realm defaultInstance = Realm.getDefaultInstance();
    List<CategoryModel> title = defaultInstance.where(CategoryModel.class).findAllSorted("title");

    defaultInstance.close();
    return title;
}

private void writeData(List<CategoryModel> categoryModels) {
    try {
        Realm defaultInstance = Realm.getDefaultInstance();
        defaultInstance.executeTransactionAsync(realm -> realm.insertOrUpdate(categoryModels));
        defaultInstance.close();
    } finally {
        getView().notifyActivity(categoryModels);
    }
}

You're supposed to have something like

private Observable<List<CategoryModel>> readData() { // Flowable with LATEST might be better.
    return io.reactivex.Observable.create(new ObservableOnSubscribe<List<CategoryModel>>() {
        @Override
        public void subscribe(ObservableEmitter<List<CategoryModel>> emitter)
                throws Exception {
            final Realm observableRealm = Realm.getDefaultInstance();
            final RealmResults<CategoryModel> results = observableRealm.where(CategoryModel.class).findAllSortedAsync("title");
            final RealmChangeListener<RealmResults<CategoryModel>> listener = results -> {
                if(!emitter.isDisposed() && results.isLoaded()) {
                    emitter.onNext(results);
                }
            };

            emitter.setDisposable(Disposables.fromRunnable(() -> {
                if(results.isValid()) {
                    results.removeChangeListener(listener);
                }
                observableRealm.close();
            }));
            results.addChangeListener(listener);
        }
    }).subscribeOn(AndroidSchedulers.mainThread())
            .unsubscribeOn(AndroidSchedulers.mainThread());
}

private void setSubscription() {
    mSubscribe = readData()
            .doOnNext((list) -> {
                if(list.isEmpty()) {
                    Single.fromCallable(() -> this::createRequest)
                            .subscribeOn(Schedulers.io())
                            .subscribe((data) -> {
                                writeData(data);
                            });
                }
            }).subscribe(data -> {
                if(!data.isEmpty()) {
                    getView().hideLoading();
                    getView().notifyActivity(data);
                }
            }, throwable -> {
                throwable.printStackTrace();
            });
}

private void writeData(List<CategoryModel> categoryModels) {
    try(Realm r = Realm.getDefaultInstance()) {
        r.executeTransaction(realm -> realm.insertOrUpdate(categoryModels));
    }
}

void unsubscribe() {
    mSubscribe.dispose();
    mSubscribe = null;
}

This way (if I didn't mess anything up), you end up with the reactive data layer described here and here, except without the additional overhead of mapping out the entire results.

EDIT:

Since Realm 4.0, it is possible to expose a RealmResults directly as a Flowable (on the UI thread, or background looper thread).

public Flowable<List<MyObject>> getLiveResults() {
    try(Realm realm = Realm.getDefaultInstance()) {
        return realm.where(MyObject.class) 
                    .findAllAsync()
                    .asFlowable()
                    .filter(RealmResults::isLoaded);
    }
}