Using Realm with RxJava

suggest change

For queries, Realm provides the realmResults.asObservable() method. Observing results is only possible on looper threads (typically the UI thread).

For this to work, your configuration must contain the following

realmConfiguration = new RealmConfiguration.Builder(context)       //
                          .rxFactory(new RealmObservableFactory()) //
                             //...
                          .build();

Afterwards, you can use your results as an observable.

Observable<RealmResults<SomeObject>> observable = results.asObservable();

For asynchronous queries, you should filter the results by isLoaded(), so that you receive an event only when the query has been executed. This filter() is not needed for synchronous queries (isLoaded() always returns true on sync queries).

Subscription subscription = RxTextView.textChanges(editText).switchMap(charSequence -> 
    realm.where(SomeObject.class)
         .contains("searchField", charSequence.toString(), Case.INSENSITIVE)
         .findAllAsync()
         .asObservable())
.filter(RealmResults::isLoaded) //
.subscribe(objects -> adapter.updateData(objects));

For writes, you should either use the executeTransactionAsync() method, or open a Realm instance on the background thread, execute the transaction synchronously, then close the Realm instance.

public Subscription loadObjectsFromNetwork() {
    return objectApi.getObjects()
        .subscribeOn(Schedulers.io())
        .subscribe(response -> {
            try(Realm realmInstance = Realm.getDefaultInstance()) {
                realmInstance.executeTransaction(realm -> realm.insertOrUpdate(response.objects));
            }
        });
}

Found a mistake? Have a question or improvement idea? Let me know.

Table Of Contents