A persistent queue implemented using realm for persistence.

RealmQueue.java

import android.content.Context;
import android.support.annotation.NonNull;
import android.support.annotation.VisibleForTesting;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;

import io.realm.Realm;
import io.realm.RealmConfiguration;
import io.realm.RealmResults;

/**
 * Created by talview23 on 26/7/16.
 */
public class RealmQueue implements Queue<String>, Closeable {
    private Realm mRealm;
    private AtomicInteger mPrimaryKeyValue;
    private boolean mClosed = false;
    private OnRealmQueueOnAddedListener mListener;

    public RealmQueue(Context context, OnRealmQueueOnAddedListener listener) {
        mRealm = Realm.getInstance(new RealmConfiguration.Builder(context.getApplicationContext()).build());
        Number maxPrimaryKeyValue = mRealm.where(RealmString.class)
                .max(RealmString.FIELD_NAME_ID);
        mPrimaryKeyValue = new AtomicInteger(maxPrimaryKeyValue == null ? 0 : maxPrimaryKeyValue.intValue());
        mListener = listener;
    }

    @VisibleForTesting
    RealmQueue(Realm realm) {
        mRealm = realm;
        Number maxPrimaryKeyValue = mRealm.where(RealmString.class)
                .max(RealmString.FIELD_NAME_ID);
        mPrimaryKeyValue = new AtomicInteger(maxPrimaryKeyValue == null ? 0 : maxPrimaryKeyValue.intValue());
    }

    private void checkIfClosed() {
        if (mClosed)
            throw new IllegalStateException("This queue is already closed, please create a new one");
    }

    @Override
    public boolean add(String string) {
        checkIfClosed();
        RealmString realmString = new RealmString(string);
        try {
            mRealm.beginTransaction();
            realmString.setId(mPrimaryKeyValue.incrementAndGet());
            mRealm.insert(realmString);
            mRealm.commitTransaction();
            if (mListener != null)
                mListener.onAdded(string);
            return true;
        } catch (IllegalStateException IEx) {
            throw new IllegalStateException("This method needs to run on the same thread as this" +
                    " object was created in");
        } catch (Exception e) {
            // todo: add crashlytics here.
            return false;
        }
    }

    @Override
    public boolean addAll(@NonNull Collection<? extends String> collection) {
        checkIfClosed();
        ArrayList<RealmString> realmStrings = new ArrayList<>(collection.size());
        for (String string : collection) {
            RealmString realmString = new RealmString(string);
            realmString.setId(mPrimaryKeyValue.incrementAndGet());
            realmStrings.add(realmString);
        }
        try {
            mRealm.beginTransaction();
            for (RealmString string : realmStrings) {
                mRealm.insert(string);
                if (mListener != null)
                    mListener.onAdded(string.getJsonData());
            }
            mRealm.commitTransaction();
            return true;
        } catch (IllegalStateException IEx) {
            throw new IllegalStateException("This method needs to run on the same thread as this" +
                    " object was created in");
        } catch (Exception e) {
            // todo: add crashlytics here.
            return false;
        }
    }

    @Override
    public void clear() {
        checkIfClosed();
        try {
            mRealm.beginTransaction();
            mRealm.deleteAll();
            mRealm.commitTransaction();
        } catch (IllegalStateException iEx) {
            throw new IllegalStateException("This method needs to run on the same thread as this" +
                    " object was created in");
        }
    }

    @Override
    public boolean contains(Object object) {
        checkIfClosed();
        if (!(object instanceof String)) {
            throw new IllegalArgumentException("Only String is supported");
        }
        String string = (String) object;
        return mRealm.where(RealmString.class).equalTo(RealmString.FIELD_VALUE, string).count() > 0;
    }

    @Override
    public boolean containsAll(@NonNull Collection<?> collection) {
        checkIfClosed();
        for (Object item : collection) {
            if (!contains(item)) {
                return false;
            }
        }
        return true;
    }

    @Override
    public boolean isEmpty() {
        checkIfClosed();
        return mRealm.where(RealmString.class).count() <= 0;
    }

    @NonNull
    @Override
    public Iterator<String> iterator() {
        checkIfClosed();
        LinkedList<String> list = new LinkedList<>();
        RealmResults<RealmString> results = mRealm.where(RealmString.class).findAll();
        for (RealmString realmString : results) {
            list.add(realmString.getJsonData());
        }
        return list.iterator();
    }

    @Override
    public boolean remove(Object object) {
        checkIfClosed();
        if (!(object instanceof String)) {
            throw new IllegalArgumentException("Only String data type supported");
        }
        String realmString = (String) object;
        mRealm.beginTransaction();
        mRealm.where(RealmString.class).equalTo(RealmString.FIELD_VALUE, realmString)
                .findFirst().deleteFromRealm();
        mRealm.commitTransaction();
        return true;
    }

    @Override
    public boolean removeAll(@NonNull Collection<?> collection) {
        checkIfClosed();
        mRealm.beginTransaction();
        for (Object object : collection) {
            if (!(object instanceof String)) {
                throw new IllegalArgumentException("Only String data type supported");
            }
            String realmString = (String) object;
            mRealm.where(RealmString.class).equalTo(RealmString.FIELD_VALUE, realmString)
                    .findFirst().deleteFromRealm();
        }
        mRealm.commitTransaction();
        return true;
    }

    @Override
    public boolean retainAll(@NonNull Collection<?> collection) {
        checkIfClosed();
        mRealm.beginTransaction();
        RealmResults<RealmString> results = mRealm.where(RealmString.class).findAll();
        for (RealmString realmStr : results) {
            if (!collection.contains(realmStr.getJsonData())) {
                realmStr.deleteFromRealm();
            }
        }
        return true;
    }

    @Override
    public int size() {
        checkIfClosed();
        return (int) mRealm.where(RealmString.class).count();
    }

    @NonNull
    @Override
    public Object[] toArray() {
        checkIfClosed();
        RealmResults<RealmString> results = mRealm.where(RealmString.class).findAll();
        ArrayList<String> returnList = new ArrayList<>(results.size());
        for (RealmString realmString : results) {
            returnList.add(realmString.getJsonData());
        }
        return returnList.toArray();
    }

    @NonNull
    @Override
    public <T> T[] toArray(@NonNull T[] array) {
        checkIfClosed();
        RealmResults<RealmString> results = mRealm.where(RealmString.class).findAll();
        ArrayList<String> returnList = new ArrayList<>(results.size());
        for (RealmString realmString : results) {
            returnList.add(realmString.getJsonData());
        }
        //noinspection SuspiciousToArrayCall
        return returnList.toArray(array);
    }

    @Override
    public boolean offer(String string) {
        checkIfClosed();
        RealmString realmString = new RealmString(string);
        mRealm.beginTransaction();
        realmString.setId(mPrimaryKeyValue.incrementAndGet());
        mRealm.insert(realmString);
        mRealm.commitTransaction();
        if (mListener != null)
            mListener.onAdded(string);
        return true;
    }

    @Override
    public String remove() {
        checkIfClosed();
        if (isEmpty())
            throw new IllegalStateException("Queue is empty");
        return removeHeadAndReturn();
    }

    private String removeHeadAndReturn() {
        mRealm.beginTransaction();
        RealmString item = mRealm.where(RealmString.class)
                .equalTo(RealmString.FIELD_NAME_ID, mRealm.where(RealmString.class)
                        .min(RealmString.FIELD_NAME_ID).intValue()).findFirst();
        String returnData = item.getJsonData();
        item.deleteFromRealm();
        mRealm.commitTransaction();
        return returnData;
    }

    @Override
    public String poll() {
        checkIfClosed();
        if (isEmpty())
            return null;
        return removeHeadAndReturn();
    }

    @Override
    public String element() {
        checkIfClosed();
        if (isEmpty())
            throw new IllegalStateException("Queue is empty");
        RealmString item = mRealm.where(RealmString.class)
                .equalTo(RealmString.FIELD_NAME_ID, mRealm.where(RealmString.class)
                        .min(RealmString.FIELD_NAME_ID).intValue()).findFirst();
        return item.getJsonData();
    }

    @Override
    public String peek() {
        checkIfClosed();
        if (isEmpty())
            return null;
        RealmString item = mRealm.where(RealmString.class)
                .equalTo(RealmString.FIELD_NAME_ID, mRealm.where(RealmString.class)
                        .min(RealmString.FIELD_NAME_ID).intValue()).findFirst();
        return item.getJsonData();
    }

    @Override
    public void close() throws IOException {
        checkIfClosed();
        mRealm.close();
        mClosed = true;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        for (String string : this) {
            sb.append(string);
            sb.append("\n");
        }
        return sb.toString();
    }
}