RealmManagerService.java

package io.gojunto.junto.services;

import io.realm.Realm;
import io.realm.RealmObject;
import io.realm.RealmResults;

/**
 * Created by danielchristopher on 5/27/17.
 */

public class RealmManagerService {
    private ThreadLocal<Realm> realms;

    public RealmManagerService() {
        realms = new ThreadLocal<>();
    }

    public Realm openRealm() {
        Realm realm = realms.get();

        if(realm != null && !realm.isClosed()) {
            throw new IllegalStateException("Realm is already open");
        }

        realm = Realm.getDefaultInstance();
        realms.set(realm);
        return realm;
    }

    public Realm getRealm() {
        Realm realm = realms.get();

        if(realm == null) {
            // No realm is open, open one
            return openRealm();
        }

        return realm;
    }

    public void closeRealm() {
        Realm realm = realms.get();

        if(realm == null) {
            throw new IllegalStateException("No Realm found to close");
        }

        if(!realm.isClosed()) {
            realm.close();
        }

        realms.set(null);
    }
}

RealmRepository.java

package io.gojunto.junto.repositories;

import java.util.AbstractList;

import io.gojunto.junto.services.RealmManagerService;
import io.reactivex.Completable;
import io.reactivex.Observable;
import io.realm.Realm;
import io.realm.RealmModel;
import io.realm.RealmObject;
import io.realm.RealmQuery;
import io.realm.RealmResults;

/**
 * Created by danielchristopher on 5/31/17.
 */

public class RealmRepository {

    public interface Transaction {
        void execute(Database db);
    }

    public interface Database {
        <E extends RealmObject> E createObject(Class<E> cls);
    }

    public class Results<T extends RealmObject> extends AbstractList<T> {
        private final RealmResults<T> results;

        protected Results(RealmResults<T> results) {
            this.results = results;
        }

        public boolean isLoaded() {
            return results.isLoaded();
        }

        @Override
        public T get(int i) {
            return results.get(i);
        }

        @Override
        public int size() {
            return results.size();
        }
    }

    public interface Query<T extends RealmObject> {
        Observable<Results<T>> findAll();
        T findFirst();
    }

    private final RealmManagerService realmManager;

    protected RealmRepository(RealmManagerService realmManager) {
        this.realmManager = realmManager;
    }

    public <T extends RealmObject> Query<T> where(Class<T> cls) {
        return wrapQuery(realmManager.getRealm().where(cls));
    }

    public Completable executeTransactionAsync(RealmRepository.Transaction transaction) {
        return Completable.create(observer -> {
            realmManager.getRealm().executeTransactionAsync(realm -> {
                transaction.execute(wrapRealm(realm));
                observer.onComplete();
            }, error -> {
                observer.onError(error);
                observer.onComplete();
            });
        });
    }

    public Completable executeTransaction(RealmRepository.Transaction transaction) {
        realmManager.getRealm().executeTransaction(realm -> {
            transaction.execute(wrapRealm(realm));
        });

        return Completable.complete();
    }

    private Database wrapRealm(Realm realm) {
        return new Database() {
            @Override
            public <E extends RealmObject> E createObject(Class<E> cls) {
                return realm.createObject(cls);
            }
        };
    }

    private <E extends RealmObject> Query<E> wrapQuery(RealmQuery<E> query) {
        return new Query<E>() {
            @Override
            public Observable<Results<E>> findAll() {
                return wrapResults(query.findAll().asObservable());
            }

            @Override
            public E findFirst() {
                return query.findFirst();
            }
        };
    }

    private <E extends RealmObject> Observable<Results<E>> wrapResults(Observable<RealmResults<E>> results) {
        return results.map(Results::new);
    }
}