RealmUtils.java

public final class RealmUtils {

    private RealmUtils() {
    }

    public static <T extends RealmObject> Observable<List<T>> createObservableList(Func1<Realm, RealmResults<T>>
            query) {
        final HandlerThread dbHandler = createDbHandler();
        final Scheduler scheduler = AndroidSchedulers.from(dbHandler.getLooper());
        final AtomicReference<Realm> realmReference = new AtomicReference<>(null);
        return Observable.defer(() -> {
            final Realm realm = Realm.getDefaultInstance();
            realmReference.set(realm);
            return query.call(realm).asObservable();
        })
                .filter(results -> results.isLoaded() && results.isValid())
                .map(results -> realmReference.get().copyFromRealm(results))
                .subscribeOn(scheduler)
                .unsubscribeOn(scheduler)
                .doOnUnsubscribe(() -> unsubscribeOnDbThread(dbHandler, realmReference.get()));
    }

    public static <T extends RealmObject> Observable<T> createObservable(Func1<Realm, T> query) {
        final HandlerThread dbHandler = createDbHandler();
        final Scheduler scheduler = AndroidSchedulers.from(dbHandler.getLooper());
        final AtomicReference<Realm> realmReference = new AtomicReference<>(null);
        return Observable.defer(() -> {
            final Realm realm = Realm.getDefaultInstance();
            realmReference.set(realm);
            final T object = query.call(realm);
            return object == null ? Observable.empty() : object.asObservable();
        })
                .filter(realmObject -> realmObject.isLoaded() && realmObject.isValid())
                .map(realmObject -> {
                    //noinspection unchecked
                    return realmReference.get().copyFromRealm((T) realmObject);
                })
                .subscribeOn(scheduler)
                .unsubscribeOn(scheduler)
                .doOnUnsubscribe(() -> unsubscribeOnDbThread(dbHandler, realmReference.get()));
    }

    public static <T extends RealmObject> Single<List<T>> createSingleList(Func1<Realm, RealmResults<T>> query) {
        return Single.fromEmitter(listEmitter -> {
            final Realm realm = Realm.getDefaultInstance();
            final RealmResults<T> realmResults = query.call(realm);
            listEmitter.onSuccess(realm.copyFromRealm(realmResults));
            listEmitter.setCancellation(realm::close);
        });
    }

    public static <T extends RealmObject> Single<T> createSingle(Func1<Realm, T> query) {
        return Single.fromEmitter(emitter -> {
            final Realm realm = Realm.getDefaultInstance();
            final T result = query.call(realm);
            if (result != null) {
                emitter.onSuccess(realm.copyFromRealm(result));
            } else {
                emitter.onSuccess(null);
            }
            emitter.setCancellation(realm::close);
        });
    }

    public static void doTransactional(Action1<Realm> transaction) {
        try (Realm realm = Realm.getDefaultInstance()) {
            realm.executeTransaction(transaction::call);
        }
    }

    private static HandlerThread createDbHandler() {
        final HandlerThread handlerThread = new HandlerThread("RealmReadThread", Process.THREAD_PRIORITY_BACKGROUND);
        handlerThread.start();
        return handlerThread;
    }

    private static void unsubscribeOnDbThread(HandlerThread handlerThread, Realm realm) {
        final Handler handler = new Handler(handlerThread.getLooper());
        handler.post(() -> {
            realm.close();
            if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.JELLY_BEAN_MR2) {
                handlerThread.quitSafely();
            } else {
                handlerThread.quit();
            }
        });
    }
}