public class Bucket<T extends Syncable> {
    
    public interface Channel {
        Change queueLocalChange(String simperiumKey);
        Change queueLocalDeletion(Syncable object);
        void log(int level, CharSequence message);
        void start();
        void stop();
        void reset();
        boolean isIdle();
        void getRevisions(String key, int sinceVersion, int maxVersionCount, RevisionsRequestCallbacks callbacks);
    }

    public interface OnBeforeUpdateObjectListener<T extends Syncable> {
        void onBeforeUpdateObject(Bucket<T> bucket, T object);
    }

    public interface OnSaveObjectListener<T extends Syncable> {
        void onSaveObject(Bucket<T> bucket, T object);
    }

    public interface OnDeleteObjectListener<T extends Syncable> {
        void onDeleteObject(Bucket<T> bucket, T object);
    }

    public interface OnNetworkChangeListener<T extends Syncable> {
        void onNetworkChange(Bucket<T> bucket, ChangeType type, String key);
    }

    public interface OnSyncObjectListener<T extends Syncable> {
        void onSyncObject(Bucket<T> bucket, String key);
    }

    public interface OnLocalQueueChangeListener<T extends Syncable> {
        void onLocalQueueChange(Bucket<T> bucket, Set<String> queuedObjects);
    }

    public interface Listener<T extends Syncable> extends
        OnSaveObjectListener<T>, OnDeleteObjectListener<T>,
        OnNetworkChangeListener<T>, OnBeforeUpdateObjectListener<T>,
        OnSyncObjectListener<T>, OnLocalQueueChangeListener<T> {
            // implements all listener methods
    }

    public interface RevisionsRequestCallbacks<T extends Syncable> {
        void onComplete(Map<Integer, T> revisions);
        void onRevision(String key, int version, JSONObject object);
        void onError(Throwable exception);
    }

    public interface RevisionsRequest {
        boolean isComplete();
    }

    /**
     * A one-way exclusion lock that stores multiple keys in a Set. Designed for use with two types of threads,
     * primary and secondary, which share a key.
     *
     * Primary threads control locking by creating a key using <code>lock(E)</code> and deleting it using
     * <code>unlock(E)</code>.
     *
     * Secondary threads check whether their key is safe to use using <code>checkLocked(E)</code>. If their key is
     * present it's not safe to use it, and the threads are blocked until the key is removed from the set.
     *
     * @param <E> the type of keys maintained
     */
    public class LockSet<E> {
        private final Set<E> mKeys = new HashSet<>();

        /**
         * Create a new key, blocking secondary threads using that key until <code>unlock(E)</code> is called
         */
        public synchronized void lock(E key) {
            mKeys.add(key);
        }

        /**
         * Block calling thread until lock is released
         */
        public synchronized void checkLocked(E key) throws InterruptedException {
            while (mKeys.contains(key)) {
                wait();
            }
        }

        /**
         * Remove a key from the map, releasing the lock
         */
        public synchronized void unlock(E key){
            mKeys.remove(key);
            notify();
        }

        /**
         * Return lock status for given key
         */
        public synchronized boolean isLocked(E key){
            return mKeys.contains(key);
        }
    }

    public enum ChangeType {
        REMOVE, MODIFY, INDEX, RESET, INSERT
    }

    public static final String TAG="Simperium.Bucket";

    private static final int BACKUP_STORE_RESET_DELAY = 5000;

    // The name used for the Simperium namespace
    private String mName;
    // User provides the access token for authentication
    private User mUser;
    // The channel that provides networking and change processing.
    private Channel mChannel;
    // For storing the bucket listeners
    private Set<OnSaveObjectListener<T>> onSaveListeners =
        Collections.synchronizedSet(new HashSet<OnSaveObjectListener<T>>());
    private Set<OnDeleteObjectListener<T>> onDeleteListeners = 
        Collections.synchronizedSet(new HashSet<OnDeleteObjectListener<T>>());
    private Set<OnBeforeUpdateObjectListener<T>> onBeforeUpdateListeners =
        Collections.synchronizedSet(new HashSet<OnBeforeUpdateObjectListener<T>>());
    private Set<OnNetworkChangeListener<T>> onChangeListeners =
        Collections.synchronizedSet(new HashSet<OnNetworkChangeListener<T>>());
    private Set<OnSyncObjectListener<T>> onSyncListeners =
        Collections.synchronizedSet(new HashSet<OnSyncObjectListener<T>>());
    private Set<OnLocalQueueChangeListener<T>> onLocalQueueChangeListeners =
        Collections.synchronizedSet(new HashSet<OnLocalQueueChangeListener<T>>());

    private BucketStore<T> mStorage;
    private BucketSchema<T> mSchema;
    private GhostStorageProvider mGhostStore;
    final private Executor mExecutor;
    private final Map<String,T> mBackupStore = new TimestampHashMap<>(BACKUP_STORE_RESET_DELAY);

    private final LockSet<String> mSaveDeleteLock = new LockSet<>();

    /**
     * Represents a Simperium bucket which is a namespace where an app syncs a user's data
     * @param name the name to use for the bucket namespace
     * @param user provides a way to namespace data if a different user logs in
     */
    public Bucket(Executor executor, String name, BucketSchema<T>schema, User user,
        BucketStore<T> storage, GhostStorageProvider ghostStore)
    throws BucketNameInvalid {
        mExecutor = executor;
        mName = name;
        mUser = user;
        mStorage = storage;
        mGhostStore = ghostStore;
        mSchema = schema;
        validateBucketName(name);
    }

    public void log(int level, CharSequence message) {
        if (mChannel == null) return;
        mChannel.log(level, message);
    }

    public void log(CharSequence message) {
        log(ChannelProvider.LOG_VERBOSE, message);
    }

    /**
     * Return the instance of this bucket's schema
     */
    public BucketSchema<T> getSchema() {
        return mSchema;
    }

    /**
     * Return the user for this bucket
     */
    public User getUser() {
        return mUser;
    }

    /**
     * Cursor for bucket data
     */
    public interface ObjectCursor<T extends Syncable> extends Cursor {

        /**
         * Return the current item's siperium key
         */
        String getSimperiumKey();
        /**
         * Return the object for the current index in the cursor
         */
        T getObject();

    }

    private class BucketCursor extends CursorWrapper implements ObjectCursor<T> {

        private ObjectCursor<T> cursor;

        BucketCursor(ObjectCursor<T> cursor) {
            super(cursor);
            this.cursor = cursor;
        }

        @Override
        public String getSimperiumKey() {
            return cursor.getSimperiumKey();
        }

        @Override
        public T getObject() {
            String key = getSimperiumKey();

            T object = cursor.getObject();
            try {
                Ghost ghost = mGhostStore.getGhost(Bucket.this, key);
                object.setGhost(ghost);
            } catch (GhostMissingException e) {
                object.setGhost(new Ghost(key, 0, new JSONObject()));
            }
            object.setBucket(Bucket.this);
            return object;
        }

    }

    /**
     * A HashMap which maintains a timestamp of its last put call and only performs <code>remove(Object)</code>
     * and <code>clear()</code> operations if a long enough interval has elapsed.
     */
    public class TimestampHashMap<K,V> extends HashMap<K,V> {
        private long mTimestamp;
        private long mClearDelay;

        public TimestampHashMap(long clearDelay) {
            mClearDelay = clearDelay;
        }

        @Override
        public V put(K key, V value) {
            V result = super.put(key, value);
            mTimestamp = System.currentTimeMillis();
            return result;
        }

        @Override
        public V remove(Object key) {
            V object = null;
            if ((System.currentTimeMillis() - mTimestamp) > mClearDelay) {
                object = super.remove(key);
            }
            return object;
        }

        @Override
        public void clear() {
            if ((System.currentTimeMillis() - mTimestamp) > mClearDelay) {
                super.clear();
            }
        }
    }

    /**
     * Tell the bucket to sync changes.
     */
    public void sync(final T object) {
        final String simperiumKey = object.getSimperiumKey();
        final String objectJSON = object.getDiffableValue().toString();
        final Boolean modified = object.isModified();

        mSaveDeleteLock.lock(simperiumKey);
        mExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    mStorage.save(object, simperiumKey, objectJSON, mSchema.indexesFor(object));

                    // Save a copy in case the object is removed from storage before this modification has been processed
                    storeBackupCopy(object);

                    mChannel.queueLocalChange(simperiumKey);

                    if (modified) {
                        // Notify listeners that an object has been saved, this was
                        // triggered locally
                        notifyOnSaveListeners(object);
                    }
                } finally {
                    mSaveDeleteLock.unlock(simperiumKey);
                }
            }
        });
    }

    /**
     * Delete the object from the bucket.
     *
     * @param object the Syncable to remove from the bucket
     */
    public void remove(T object) {
        remove(object, true);
    }

    /**
     * Remove the object from the bucket. If isLocal is true, this will queue
     * an operation to sync with the Simperium service.
     *
     * @param object The Syncable to remove from the bucket
     * @param isLocal if the operation originates from this client
     */
    private void remove(final T object, final boolean isLocal) {

        mExecutor.execute(new Runnable() {
            @Override
            public void run() {
                if (isLocal) {
                    try {
                        mSaveDeleteLock.checkLocked(object.getSimperiumKey());
                    } catch (InterruptedException e) {
                        Logger.log(TAG, String.format("Delete lock interrupted for %s, did not remove from storage",
                                object.getSimperiumKey()));
                        // Proceed, but don't remove from storage
                        mChannel.queueLocalDeletion(object);
                        notifyOnDeleteListeners(object);
                        return;
                    }

                    // Put object in backup store if not already there so that queued local deletion
                    // will send change event and remove object from server and all other platforms.
                    if (mBackupStore.get(object.getSimperiumKey()) == null) {
                        mBackupStore.put(object.getSimperiumKey(), object);
                    }

                    mChannel.queueLocalDeletion(object);
                }

                mStorage.delete(object);

                if (isLocal) {
                    notifyOnDeleteListeners(object);
                }
            }
        });
    }

    /**
     * Given the key for an object in the bucket, remove it if it exists
     */
    private void removeObjectWithKey(String key)
    throws BucketObjectMissingException {
        T object = get(key);
        if (object != null) {
            // this will call onObjectRemoved on the listener
            remove(object, false);
        }
    }

    /**
     * Store a copy of the object in the backup store
     */
    private void storeBackupCopy(T object) {
        synchronized(mBackupStore) {
            if (mChannel.isIdle()) {
                // If there is no activity in the Channel, we can try to clean up obsolete backups
                mBackupStore.clear();
            }
            mBackupStore.put(object.getSimperiumKey(), object);
        }
    }

    /**
     * Get the bucket's namespace
     * @return (String) bucket's namespace
     */
    public String getName() {
        return mName;
    }

    public String getRemoteName() {
        return mSchema.getRemoteName();
    }

    public Boolean hasChangeVersion() {
        return mGhostStore.hasChangeVersion(this);
    }

    public String getChangeVersion() {
        String version = mGhostStore.getChangeVersion(this);
        if (version == null) {
            version = "";
        }
        return version;
    }

    public void indexComplete(String changeVersion) {
        setChangeVersion(changeVersion);
        notifyOnNetworkChangeListeners(ChangeType.INDEX);
    }

    public void setChangeVersion(String version) {
        mGhostStore.setChangeVersion(this, version);
    }

    // starts tracking the object
    /**
     * Add an object to the bucket so simperium can start syncing it. Must
     * conform to the Diffable interface. So simperium can diff/apply patches.
     *
     */
    public void add(T object) {
        if (!object.getBucket().equals(this)) {
            object.setBucket(this);
        }
    }

    protected T buildObject(String key, JSONObject properties) {
        return buildObject(new Ghost(key, 0, properties));
    }


    protected T buildObject(String key) {
        return buildObject(key, new JSONObject());
    }

    protected T buildObject(Ghost ghost) {
        T object = mSchema.buildWithDefaults(ghost.getSimperiumKey(), JSONDiff.deepCopy(ghost.getDiffableValue()));
        object.setGhost(ghost);
        object.setBucket(this);
        return object;
    }

    public int count() {
        return count(query());
    }

    public int count(Query<T> query) {
        return mStorage.count(query);
    }

    /**
     * Find all objects
     */
    public ObjectCursor<T> allObjects() {
        return new BucketCursor(mStorage.all());
    }

    /**
     * Search using a query
     */
    public ObjectCursor<T> searchObjects(Query<T> query) {
        return new BucketCursor(mStorage.search(query));
    }

    /**
     * Support cancelation
     */
    /**
     * Build a query for this object
     */
    public Query<T> query() {
        return new Query<T>(this);
    }

    /**
     * Get a single object object that matches key
     */
    public T get(String key) throws BucketObjectMissingException {
        // Datastore constructs the object for us
        Ghost ghost;
        try {
            ghost = mGhostStore.getGhost(this, key);
        } catch (GhostMissingException e) {
            throw(new BucketObjectMissingException(String.format("Bucket %s does not have object %s", getName(), key)));
        }
        T object = mStorage.get(key);
        if (object == null) {
            throw(new BucketObjectMissingException(String.format("Storage provider for bucket:%s did not have object %s", getName(), key)));
        }
        Logger.log(TAG, String.format("Fetched ghost for %s %s", key, ghost));
        object.setBucket(this);
        object.setGhost(ghost);
        updateBackupStoreGhost(ghost);
        return object;
    }

    /**
     * Get an object by its key, checking the backup store if the object has been removed from the persistent store
     */
    public T getObjectOrBackup(String key) throws BucketObjectMissingException {
        T object;
        try {
            object = get(key);
        } catch (BucketObjectMissingException e) {
            // If the object has been removed from the persistent store, check the backup store
            object = mBackupStore.get(key);
            if (object == null) {
                throw(new BucketObjectMissingException(String.format(
                        "Storage provider for bucket:%s did not have object %s and there was no stored backup",
                        getName(), key)));
            }
            Logger.log(TAG, String.format("Fetched backup copy for %s", key));
        }
        return object;
    }

    /**
     * Get an object by its key, should we throw an error if the object isn't
     * there?
     */
    public T getObject(String uuid) throws BucketObjectMissingException {
        return get(uuid);
    }

    /**
     * Returns a new object tracked by this bucket
     */
    public T newObject() {
        try {
            return newObject(uuid());
        } catch (BucketObjectNameInvalid e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * Returns a new object with the given uuid
     * return null if the uuid exists?
     */
    public T newObject(String key)
    throws BucketObjectNameInvalid {
        return insertObject(key, new JSONObject());
    }

    /**
     * 
     */
    public T insertObject(String key, JSONObject properties)
    throws BucketObjectNameInvalid {
        if (key == null)
            throw new BucketObjectNameInvalid(null);

        String name = key.trim();
        validateObjectName(name);
        T object = buildObject(name, properties);
        object.setBucket(this);
        Ghost ghost = new Ghost(name, 0, new JSONObject());
        object.setGhost(ghost);
        mGhostStore.saveGhost(this, ghost);
        return object;
    }

    /**
     * Add object from new ghost data, no corresponding change version so this
     * came from an index request
     */
    protected void addObjectWithGhost(final Ghost ghost) {
        addObjectWithGhost(ghost, null);
    }

    /**
     * 
     */
    protected void addObjectWithGhost(final Ghost ghost, final Runnable runnable) {
        mExecutor.execute(new Runnable() {

            @Override
            public void run() {
                mGhostStore.saveGhost(Bucket.this, ghost);
                T object = buildObject(ghost);
                addObject(object);

                if (runnable != null) {
                    runnable.run();
                }

            }

        });
    }

    /**
     * Update the ghost data
     */
    protected void updateObjectWithGhost(final Ghost ghost) {
        mGhostStore.saveGhost(Bucket.this, ghost);
        T object = mSchema.build(ghost.getSimperiumKey(), ghost.getDiffableValue());
        updateObject(object);
    }

    protected void updateGhost(final Ghost ghost, final Runnable complete) {
        mExecutor.execute(new Runnable() {

            @Override
            public void run() {
                // find the object
                try {
                    T object = get(ghost.getSimperiumKey());
                    if (object.isModified()) {
                        // Attempt to merge local changes with the new ghost
                        Ghost localGhost = object.getGhost();
                        JSONObject localGhostProperties = localGhost.getDiffableValue();

                        // Get diff of local ghost vs. local object
                        JSONObject localModifications;
                        try {
                            localModifications = JSONDiff.diff(localGhostProperties, object.getDiffableValue());
                        } catch (JSONException e) {
                            localModifications = new JSONObject();
                        }

                        // Get diff of remote ghost vs. local ghost
                        JSONObject remoteModifications;
                        try {
                            remoteModifications = JSONDiff.diff(localGhostProperties, ghost.getDiffableValue());
                        } catch (JSONException e) {
                            remoteModifications = new JSONObject();
                        }

                        try {
                            JSONObject localPatch = localModifications.getJSONObject(JSONDiff.DIFF_VALUE_KEY);
                            JSONObject remotePatch = remoteModifications.getJSONObject(JSONDiff.DIFF_VALUE_KEY);

                            JSONObject transformedDiff = JSONDiff.transform(localPatch, remotePatch, localGhostProperties);
                            JSONObject updatedProperties = JSONDiff.deepCopy(ghost.getDiffableValue());
                            updatedProperties = JSONDiff.apply(updatedProperties, transformedDiff);

                            mSchema.update(object, updatedProperties);
                            object.setGhost(ghost);
                            updateObject(object);
                        } catch (JSONException | IllegalArgumentException e) {
                            // Could not apply patch, update from the ghost
                            updateObjectWithGhost(ghost);
                        }
                    } else {
                        // Apply the new ghost to the unmodified local object
                        updateObjectWithGhost(ghost);
                    }

                    // Notify listeners that the object has changed
                    notifyOnNetworkChangeListeners(ChangeType.MODIFY, object.getSimperiumKey());
                } catch (BucketObjectMissingException e) {
                    // The object doesn't exist, insert the new object
                    updateObjectWithGhost(ghost);
                }

                if (complete != null) {
                    complete.run();
                }
            }

        });
    }

    /**
     * Update the ghost of an object in the backup store
     */
    private void updateBackupStoreGhost(Ghost ghost) {
        T object = mBackupStore.get(ghost.getSimperiumKey());
        if (object != null) {
            object.setGhost(ghost);
        }
    }

    public Ghost getGhost(String key) throws GhostMissingException {
        return mGhostStore.getGhost(this, key);
    }
    /**
     * Add a new object with corresponding change version
     */
    protected void addObject(String changeVersion, T object) {
        addObject(object);
        setChangeVersion(changeVersion);
    }
    /**
     * Adds a new object to the bucket
     */
    protected void addObject(T object) {
        if (object.getGhost() == null) {
            object.setGhost(new Ghost(object.getSimperiumKey()));
        }

        object.setBucket(this);
        JSONObject objectJSON = object.getDiffableValue();
        mStorage.save(object, object.getSimperiumKey(), objectJSON.toString(), mSchema.indexesFor(object));
        // notify listeners that an object has been added
    }

    /**
     * Updates an existing object
     */
    protected void updateObject(T object) {
        object.setBucket(this);

        String json = object.getDiffableValue().toString();
        mStorage.save(object, object.getSimperiumKey(), json, mSchema.indexesFor(object));
    }

    /**
     *
     */
    protected void updateObject(String changeVersion, T object) {
        updateObject(object);
        setChangeVersion(changeVersion);
    }

    public void addListener(Listener<T> listener) {
        addOnSaveObjectListener(listener);
        addOnBeforeUpdateObjectListener(listener);
        addOnDeleteObjectListener(listener);
        addOnNetworkChangeListener(listener);
        addOnSyncObjectListener(listener);
        addOnLocalQueueChangeListener(listener);
    }

    public void removeListener(Listener<T> listener) {
        removeOnSaveObjectListener(listener);
        removeOnBeforeUpdateObjectListener(listener);
        removeOnDeleteObjectListener(listener);
        removeOnNetworkChangeListener(listener);
        removeOnSyncObjectListener(listener);
        removeOnLocalQueueChangeListener(listener);
    }

    public void addOnSaveObjectListener(OnSaveObjectListener<T> listener) {
        onSaveListeners.add(listener);
    }

    public void removeOnSaveObjectListener(OnSaveObjectListener<T> listener) {
        onSaveListeners.remove(listener);
    }

    public void addOnDeleteObjectListener(OnDeleteObjectListener<T> listener) {
        onDeleteListeners.add(listener);
    }

    public void removeOnDeleteObjectListener(OnDeleteObjectListener<T> listener) {
        onDeleteListeners.remove(listener);
    }

    public void addOnNetworkChangeListener(OnNetworkChangeListener<T> listener) {
        onChangeListeners.add(listener);
    }

    public void removeOnNetworkChangeListener(OnNetworkChangeListener<T> listener) {
        onChangeListeners.remove(listener);
    }

    public void addOnBeforeUpdateObjectListener(OnBeforeUpdateObjectListener<T> listener) {
        onBeforeUpdateListeners.add(listener);
    }

    public void removeOnBeforeUpdateObjectListener(OnBeforeUpdateObjectListener<T> listener) {
        onBeforeUpdateListeners.remove(listener);
    }

    public void addOnSyncObjectListener(OnSyncObjectListener<T> listener) {
        onSyncListeners.add(listener);
    }

    public void removeOnSyncObjectListener(OnSyncObjectListener<T> listener) {
        onSyncListeners.remove(listener);
    }

    public void addOnLocalQueueChangeListener(OnLocalQueueChangeListener<T> listener) {
        onLocalQueueChangeListeners.add(listener);
    }

    public void removeOnLocalQueueChangeListener(OnLocalQueueChangeListener<T> listener) {
        onLocalQueueChangeListeners.remove(listener);
    }

    private <L> void notifyListeners(Set<L> listeners, Consumer<L> annunciator) {
        for (L listener : new HashSet<>(listeners)) {
            try {
                annunciator.accept(listener);
            } catch (Exception e) {
                Logger.log(TAG, String.format(
                    "Listener failed %s %s",
                    listener.getClass().getSimpleName(),
                    listener
                ), e);
            }
        }
    }

    public void notifyOnSaveListeners(final T object) {
        notifyListeners(onSaveListeners, new Consumer<OnSaveObjectListener<T>>() {
            @Override
            public void accept(OnSaveObjectListener<T> listener) {
                listener.onSaveObject(Bucket.this, object);
            }
        });
    }

    public void notifyOnDeleteListeners(final T object) {
        notifyListeners(onDeleteListeners, new Consumer<OnDeleteObjectListener<T>>() {
            @Override
            public void accept(OnDeleteObjectListener<T> listener) {
                listener.onDeleteObject(Bucket.this, object);
            }
        });
    }

    public void notifyOnBeforeUpdateObjectListeners(final T object) {
        notifyListeners(onBeforeUpdateListeners, new Consumer<OnBeforeUpdateObjectListener<T>>() {
            @Override
            public void accept(OnBeforeUpdateObjectListener<T> listener) {
                listener.onBeforeUpdateObject(Bucket.this, object);
            }
        });
    }

    public void notifyOnNetworkChangeListeners(ChangeType type) {
        notifyOnNetworkChangeListeners(type, null);
    }

    public void notifyOnNetworkChangeListeners(final ChangeType type, final String key) {
        notifyListeners(onChangeListeners, new Consumer<OnNetworkChangeListener<T>>() {
            @Override
            public void accept(OnNetworkChangeListener<T> listener) {
                listener.onNetworkChange(Bucket.this, type, key);
            }
        });
    }

    public void notifyOnSyncObjectListeners(final String key) {
        notifyListeners(onSyncListeners, new Consumer<OnSyncObjectListener<T>>() {
            @Override
            public void accept(OnSyncObjectListener<T> listener) {
                listener.onSyncObject(Bucket.this, key);
            }
        });
    }

    public void notifyOnLocalQueueChangeListeners(final Set<String> keys) {
        notifyListeners(onLocalQueueChangeListeners, new Consumer<OnLocalQueueChangeListener<T>>() {
            @Override
            public void accept(OnLocalQueueChangeListener<T> listener) {
                listener.onLocalQueueChange(Bucket.this, keys);
            }
        });
    }

    public void setChannel(Channel channel) {
        mChannel = channel;
    }

    /**
     * Initialize the bucket to start tracking changes.
     */
    public void start() {
        mChannel.start();
    }

    public void stop() {
        mChannel.stop();
    }


    public void reset() {
        mStorage.reset();
        // Clear the ghost store
        mGhostStore.resetBucket(this);
        mChannel.reset();
        stop();

        notifyOnNetworkChangeListeners(ChangeType.RESET);
    }
    /**
     * Does bucket have at least the requested version?
     */
    public Boolean containsKey(String key) {
        return mGhostStore.hasGhost(this, key);
    }
    /**
     * Ask storage if it has at least the requested version or newer
     */
    public Boolean hasKeyVersion(String key, Integer version) {
        try {
            Ghost ghost = mGhostStore.getGhost(this, key);
            return ghost.getVersion().equals(version);
        } catch (GhostMissingException e) {
            // we don't have the ghost
            return false;
        }
    }
    /**
     * Which version of the key do we have
     */
    public Integer getKeyVersion(String key) throws GhostMissingException {
        Ghost ghost = mGhostStore.getGhost(this, key);
        return ghost.getVersion();
    }

    /**
     * Submit a Runnable to this Bucket's executor
     */
    public void executeAsync(Runnable task) {
        mExecutor.execute(task);
    }

    public String uuid() {
        String key;
        do {
            key = Uuid.uuid();
        } while(containsKey(key));
        return key;
    }

    public Ghost acknowledgeChange(RemoteChange remoteChange, Change change)
    throws RemoteChangeInvalidException {
        Ghost ghost = null;
        if (!remoteChange.isRemoveOperation()) {
            try {
                T object = getObjectOrBackup(remoteChange.getKey());
                // apply the diff to the underyling object
                ghost = remoteChange.apply(object.getGhost());
                mGhostStore.saveGhost(this, ghost);
                // update the object's ghost
                object.setGhost(ghost);
                updateBackupStoreGhost(ghost);
            } catch (BucketObjectMissingException e) {
                throw(new RemoteChangeInvalidException(remoteChange, e));
            }
        } else {
            mGhostStore.deleteGhost(this, remoteChange.getKey());
        }
        setChangeVersion(remoteChange.getChangeVersion());
        remoteChange.setApplied();
        // TODO: remove changes don't need ghosts, need to rethink this a bit
        return ghost;
    }

    public Ghost applyRemoteChange(RemoteChange change)
    throws RemoteChangeInvalidException {
        Ghost updatedGhost = null;
        if (change.isRemoveOperation()) {
            try {
                removeObjectWithKey(change.getKey());
                mGhostStore.deleteGhost(this, change.getKey());
            } catch (BucketObjectMissingException e) {
                throw(new RemoteChangeInvalidException(change, e));
            }
        } else {
            try {
                T object;
                Boolean isNew;
                Boolean shouldUpdateObject = true;

                if (change.isAddOperation()) {
                    object = newObject(change.getKey());
                    isNew = true;
                } else {
                    object = getObjectOrBackup(change.getKey());
                    isNew = false;

                    notifyOnBeforeUpdateObjectListeners(object);
                }

                Ghost ghost = object.getGhost();
                JSONObject localModifications = null;
                JSONObject currentProperties = ghost.getDiffableValue();

                try {
                    localModifications = JSONDiff.diff(currentProperties, object.getDiffableValue());
                } catch (JSONException e) {
                    localModifications = new JSONObject();
                }

                // updates the ghost and sets it on the object
                updatedGhost = change.apply(ghost);
                JSONObject updatedProperties = JSONDiff.deepCopy(updatedGhost.getDiffableValue());

                // persist the ghost to storage
                mGhostStore.saveGhost(this, updatedGhost);
                object.setGhost(updatedGhost);
                updateBackupStoreGhost(updatedGhost);

                // allow the schema to update the object instance with the new
                if (isNew) {
                    mSchema.updateWithDefaults(object, updatedProperties);
                    addObject(object);
                } else {
                    if (localModifications != null && localModifications.length() > 0) {
                        try {
                            JSONObject incomingDiff = change.getPatch();
                            JSONObject localDiff = localModifications.getJSONObject(JSONDiff.DIFF_VALUE_KEY);

                            JSONObject transformedDiff = JSONDiff.transform(localDiff, incomingDiff, currentProperties);

                            updatedProperties = JSONDiff.apply(updatedProperties, transformedDiff);
                        } catch (JSONException | IllegalArgumentException e) {
                            // We couldn't merge the local and remote changes.
                            // Hold off on updating the object so that the local change can sync
                            shouldUpdateObject = false;
                        }
                    }

                    if (shouldUpdateObject) {
                        mSchema.update(object, updatedProperties);
                        updateObject(object);
                    }
                }

            } catch(SimperiumException e) {
                Logger.log(TAG, String.format("Unable to apply remote change %s", change), e);
                throw(new RemoteChangeInvalidException(change, e));
            }
        }
        setChangeVersion(change.getChangeVersion());
        change.setApplied();

        ChangeType type;
        if (change.isAddOperation()) {
            type = ChangeType.INSERT;
        } else if (change.isRemoveOperation()) {
            type = ChangeType.REMOVE;
        } else {
            type = ChangeType.MODIFY;
        }

        notifyOnNetworkChangeListeners(type, change.getKey());
        return updatedGhost;
    }

    static public final String BUCKET_OBJECT_NAME_REGEX = "^[a-zA-Z0-9_\\.\\-%@]{1,256}$";

    public static void validateObjectName(String name)
    throws BucketObjectNameInvalid {
        if (name == null || !name.matches(BUCKET_OBJECT_NAME_REGEX)) {
            throw new BucketObjectNameInvalid(name);
        }
    }

    static public final String BUCKET_NAME_REGEX = "^[a-zA-Z0-9_\\.\\-%]{1,64}$";

    public static void validateBucketName(String name)
    throws BucketNameInvalid {
        if (name == null || !name.matches(BUCKET_NAME_REGEX)) {
            throw new BucketNameInvalid(name);
        }
    }

    /**
     * Request revision history for an object. Called by all of the public getRevision() methods.
     */
    private void getRevisions(String key, int version, int max, final RevisionsRequestCallbacks<T> callbacks) {
        mChannel.getRevisions(key, version, max, new RevisionsRequestCallbacks() {

            @Override
            public void onComplete(Map revisions) {
                callbacks.onComplete(revisions);
            }

            @Override
            public void onRevision(String key, int version, JSONObject object) {
                callbacks.onRevision(key, version, object);
            }

            @Override
            public void onError(Throwable exception) {
                callbacks.onError(exception);
            }
        });
    }

    public void getRevisions(final T object, final int max, final RevisionsRequestCallbacks<T> callbacks) {
        mExecutor.execute(new Runnable() {
            @Override
            public void run() {
                getRevisions(object.getSimperiumKey(), object.getVersion(), max, callbacks);
            }
        });
    }

    public void getRevisions(final T object, final RevisionsRequestCallbacks<T> callbacks) {
        mExecutor.execute(new Runnable() {
            @Override
            public void run() {
                getRevisions(object.getSimperiumKey(), object.getVersion(), 0, callbacks);
            }
        });
    }

    public void getRevisions(final String key, final RevisionsRequestCallbacks<T> callbacks) {
        mExecutor.execute(new Runnable() {
            @Override
            public void run() {
                int version;
                try {
                    version = mGhostStore.getGhostVersion(Bucket.this, key);
                    getRevisions(key, version, 0, callbacks);
                } catch (GhostMissingException e) {
                    callbacks.onError(e);
                }
            }
        });
    }

    public void getRevisions(final String key, final int max, final RevisionsRequestCallbacks<T> callbacks) {
        mExecutor.execute(new Runnable() {
            @Override
            public void run() {
                int version;
                try {
                    version = mGhostStore.getGhostVersion(Bucket.this, key);
                    getRevisions(key, version, max, callbacks);
                } catch (GhostMissingException e) {
                    callbacks.onError(e);
                }
            }
        });
    }
} public abstract class BucketSchema<T extends Syncable> {

    static public final String TAG = "Simperium";

    public abstract String getRemoteName();
    public abstract T build(String key, JSONObject properties);
    public abstract void update(T object, JSONObject properties);

    public interface Indexer<S extends Syncable> {
        public List<Index> index(S object);
    }

    public static class Index {
        private String name;
        private Object value;

        public Index(String name, Object value){
            this.name = name;
            this.value = value;
        }

        public String getName(){
            return name;
        }

        public Object getValue(){
            return value;
        }

        public String toString(){
            return String.format("%s : %s", name, value);
        }
    }

    private List<Indexer<T>> indexers = Collections.synchronizedList(new ArrayList<Indexer<T>>());
    private Map<String,Object> defaultValues = new HashMap<String,Object>();
    private FullTextIndex mFullTextIndex;

    public T buildWithDefaults(String key, JSONObject properties) {
        updateDefaultValues(properties);
        return build(key, properties);
    }

    public void updateWithDefaults(T object, JSONObject properties){
        updateDefaultValues(properties);
        update(object, properties);
    }

    protected void setDefault(String key, Object value){
        defaultValues.put(key, value);
    }

    protected void updateDefaultValues(JSONObject properties){
        Set<Entry<String,Object>> defaults = defaultValues.entrySet();
        Iterator<Entry<String,Object>> iterator = defaults.iterator();
        while(iterator.hasNext()){
            Entry<String,Object> entry = iterator.next();
            String key = entry.getKey();
            if (!properties.has(key)) {
                try {
                    properties.put(key, JSONDiff.deepCopy(entry.getValue()));
                } catch (JSONException e) {
                    android.util.Log.e(TAG, "Could not assign default value " + key, e);
                }
            }
        }
    }

    public List<Index> indexesFor(T object){
        // run all of the indexers
        List<Index> indexes = new ArrayList<Index>();
        Iterator<Indexer<T>> indexerIterator = indexers.iterator();
        while(indexerIterator.hasNext()){
            indexes.addAll(indexerIterator.next().index(object));
        }
        return indexes;
    }

    public void addIndex(Indexer indexer){
        indexers.add(indexer);
    }

    public void removeIndex(Indexer indexer){
        indexers.remove(indexer);
    }

    public void autoIndex(){
        indexers.add(0, new AutoIndexer<T>());
    }

    public FullTextIndex<T> setupFullTextIndex(String ... indexNames){

        FullTextIndex.Indexer<T> indexer = new FullTextIndex.Indexer<T>() {

            @Override
            public Map<String,String> index(String[] keys, T object){
                Map<String,String> indexValues = new HashMap<String,String>(keys.length);
                JSONObject values = object.getDiffableValue();
                for (String key : keys) {
                    Object value = values.opt(key);
                    if (value != null) {
                        indexValues.put(key, value.toString());
                    }
                }

                return indexValues;
            }

        };

        return setupFullTextIndex(indexer, indexNames);
    }

    public FullTextIndex<T> setupFullTextIndex(FullTextIndex.Indexer<T> indexer, String ... indexNames){
        mFullTextIndex = new FullTextIndex<T>(indexer, indexNames);
        return mFullTextIndex;
    }

    public boolean hasFullTextIndex(){
        return mFullTextIndex != null;
    }

    public FullTextIndex<T> getFullTextIndex(){
        return mFullTextIndex;
    }

    private static class AutoIndexer<S extends Syncable> implements Indexer<S> {

        public List<Index> index(S object){

            JSONObject values = object.getDiffableValue();
            List<Index> indexes = new ArrayList<Index>(values.length());
            Iterator keys = values.keys();

            while (keys.hasNext()) {
                String key = keys.next().toString();
                try {
                    if (values.has(key))
                        addIndex(indexes, key, values.get(key));
                } catch (JSONException e) {
                    android.util.Log.e(TAG, "Unable to index " + key, e);
                }
            }

            return indexes;
        }

        private void addIndex(List<Index> indexes, String key, Object value){
            if (value instanceof JSONArray) {
                JSONArray list = (JSONArray) value;
                int length = list.length();
                for (int i = 0; i < length; i++) {
                    try {
                        addIndex(indexes, key, list.get(i));
                    } catch (JSONException e) {
                        android.util.Log.e(TAG, "Unable to index array", e);
                    }
                }
            } else {
                indexes.add(new Index(key, value));
            }
        }

    }

} public interface GhostStorageProvider {
    /**
     * Check if the store has a change version for the provided bucket
     */
    public boolean hasChangeVersion(Bucket bucket);
    /**
     * Returns wether the bucket's saved version matches the given version
     */
    public boolean hasChangeVersion(Bucket bucket, String version);
    /**
     * Returns the current change version for the bucket
     */
    public String getChangeVersion(Bucket bucket);
    /**
     * Updates the change version for the given bucket
     */
    public void setChangeVersion(Bucket bucket, String version);
    /**
     * Checks if there is a ghost for given bucket and key
     */
    public boolean hasGhost(Bucket bucket, String key);
    /**
     * Builds a ghost from the provided bucket and key
     */
    public Ghost getGhost(Bucket bucket, String key) throws GhostMissingException;
    /**
     * Get a ghost's version number
     */
    public int getGhostVersion(Bucket bucket, String key) throws GhostMissingException;
    /**
     * Saves the provided ghost to the bucket
     */
    public void saveGhost(Bucket bucket, Ghost ghost);
    /**
     * Remove the ghost with the provided key from the store
     */
    public void deleteGhost(Bucket bucket, String key);
    /**
     * Clear all ghost data and change version for the given bucket
     */
    public void resetBucket(Bucket bucket);
} public class User {
    /**
     * Applications can register a global authentication listener to get notified when a user's
     * authenticated status has changed.
     *
     *     Simperium simperium = new Simperium(appId, appSecret, appContext, new User.AuthenticationListener(){
     *       public void onAuthenticationStatusChange(User.AuthenticationStatus status){
     *          // Prompt user to log in or show they're offline
     *       }
     *     });
     */
    public interface StatusChangeListener {
        void onUserStatusChange(Status authorized);
    }
    /**
     * Determines a user's network status with Simperium:
     *   - AUTHENTICATED: user has an access token and is connected to Simperium
     *   - NOT_AUTHENTICATED: user does not have a valid access token. Create or auth the user
     *   - UKNOWN: User objects start in this state and then transitions to AUTHENTICATED or
     *             NOT_AUTHENTICATED. Also the state given when the Simperium is not reachable.
     */
    public enum Status {
        AUTHORIZED, NOT_AUTHORIZED, UNKNOWN
    }

    private String email;
    private String password;
    private String userId;
    private String accessToken;
    private Status status = Status.UNKNOWN;
    private StatusChangeListener listener;

    public User(){
        this(null, null, null);
    }
    // a user that hasn't been logged in
    public User(StatusChangeListener listener){
        this(null, null, listener);
    }

    public User(String email, StatusChangeListener listener){
        this(email, null, listener);
    }

    public User(String email, String password, StatusChangeListener listener){
        this.email = email;
        this.password = password;
        this.listener = listener;
    }

    public Status getStatus(){
        return status;
    }

    public void setStatusChangeListener(StatusChangeListener authListener){
        listener = authListener;
    }

    public StatusChangeListener getStatusChangeListener(){
        return listener;
    }

    public void setStatus(Status status){
        if (this.status != status) {
            this.status = status;
            if (this.listener != null) {
                listener.onUserStatusChange(this.status);
            }
        }
    }

    // check if we have an access token
    public boolean needsAuthorization(){
        return accessToken == null;
    }

    public boolean hasAccessToken(){
        return accessToken != null;
    }

    public void setCredentials(String email, String password){
        setEmail(email);
        setPassword(password);
    }

    public String getEmail(){
        return email;
    }

    public void setEmail(String email){
        this.email = email;
    }

    public void setPassword(String password){
        this.password = password;
    }

    public String getPassword(){
        return password;
    }

    public String getUserId(){
        return userId;
    }

    public void setUserId(String userId){
        this.userId = userId;
    }

    public String getAccessToken(){
        return accessToken;
    }

    public void setAccessToken(String token){
        this.accessToken = token;
    }

} public interface BucketStore<T extends Syncable> {

    /**
     * For initializing and performaing any necessary setup for storing
     * bucket data.
     */
    public void prepare(Bucket<T> bucket);

    /**
     * Add/Update the given object
     */
    public void save(T object, String simperiumKey, String json, List<Index> indexes);

    /**
     * Remove the given object from the storage
     */
    public void delete(T object);

    /**
     * Delete all objects from storage
     */
    public void reset();

    /**
     * Get an object with the given key
     */
    public T get(String key) throws BucketObjectMissingException;

    /**
     * All objects, returns a cursor for the given bucket
     */
    public Bucket.ObjectCursor<T> all();

    /**
     * 
     */
    public Bucket.ObjectCursor<T> search(Query<T> query);

    /**
     * Return the count for the given query
     */
    public int count(Query<T> query);

} public class Note extends BucketObject {
    
    public static class Schema extends BucketSchema<Note> {
        public static final String BUCKET_NAME="notes";

        public Schema(){
            autoIndex();
            addIndex(contentIndexer);
            addIndex(nullIndexer);
            setupFullTextIndex("tags", "content");
            setDefault("tags", new JSONArray());
            setDefault("deleted", false);
        }

        private Indexer<Note> contentIndexer = new Indexer<Note>(){
            @Override
            public List<Index> index(Note note){
                List<Index> indexes = new ArrayList<Index>(1);
                indexes.add(new Index("preview", note.getPreview()));
                return indexes;
            }
        };

        private Indexer<Note> nullIndexer = new Indexer<Note>() {
            @Override
            public List<Index> index(Note note) {
                List<Index> indexes = new ArrayList<Index>(1);
                indexes.add(new Index("null_column", null));
                return indexes;
            }
        };

        @Override
        public String getRemoteName(){
            return BUCKET_NAME;
        }

        @Override
        public Note build(String key, JSONObject properties){
            return new Note(key, properties);
        }

        @Override
        public void update(Note note, JSONObject properties){
            note.setProperties(properties);
        }
    }

    private static final String SPACE=" ";
    private StringBuilder preview;

    public Note(String key, JSONObject properties){
        super(key, properties);
    }

    public void addTag(String tag){
        JSONArray tags = (JSONArray) get("tags");
        tags.put(tag);
    }

    public void addTags(String ... newTags){
        JSONArray tags = (JSONArray) get("tags");
        for(String tag : newTags){
            tags.put(tag);
        }
    }

    public void setTitle(String title){
        put("title", title);
    }

    public String getTitle(){
        return (String) get("title");
    }

    public String getContent(){
        return (String) get("content");
    }

    public void setContent(String content){
        preview = null;
        put("content", content);
    }

    public void put(String key, Object value){
        try {
            getProperties().put(key, value);
        } catch (JSONException e) {
            // Do nothing
        }
    }

    public Object get(String key){
        try {
            return getProperties().get(key);
        } catch (JSONException e) {
            // Do nothing
            return null;
        }
    }

    public CharSequence getPreview(){
        if (preview != null) {
            return preview;
        }
        String content = getContent();
        if (content == null) {
            return "";
        }
        // just the first three lines
        preview = new StringBuilder();
        int start = 0;
        int position = -1;
        int lines = 0;
        do {
            position = content.indexOf("\n", start);
            // if there are no new lines in the whole thing, we'll just take the whole thing
            if (position == -1 && start == 0) {
                position = content.length();
            }
            if (position > start + 1) {
                int length = preview.length();
                if (length > 0) {
                    preview.append(SPACE);
                }
                preview.append(content.subSequence(start, position));
                if (length > 320) {
                    break;
                }
                lines ++;
            }
            start = position + 1;
        } while(lines < 3 && position > -1 && start > 0);
        return preview;
    }
} public class MockChannel implements Bucket.Channel {

    public static final String TAG = "Simperium.Test";

    private Bucket mBucket;
    private boolean started = false;
    private boolean idle = false;

    public boolean autoAcknowledge = true;

    public MockChannel(Bucket bucket){
        mBucket = bucket;
    }

    @Override
    public Change queueLocalDeletion(Syncable object){
        Change change = new Change(Change.OPERATION_REMOVE, mBucket.getName(), object.getSimperiumKey());
        try {
            if(started && autoAcknowledge) acknowledge(change);
        } catch (Exception e) {
            throw(new RuntimeException(e));
        }
        return change;
    }

    @Override
    public boolean isIdle() {
        return idle;
    }

    @Override
    public Change queueLocalChange(String simperiumKey) {
        Change change = new Change(Change.OPERATION_MODIFY, mBucket.getName(), simperiumKey);
        try {
            if(started && autoAcknowledge) acknowledge(change);
        } catch (Exception e) {
            throw(new RuntimeException(e));
        }
        return change;
    }

    @Override
    public void log(int level, CharSequence message) {
        Log.d(TAG, String.format("Remote log (%d): %s", level, message));
    }

    @Override
    public void reset(){}

    @Override
    public void start(){
        started = true;
    }

    @Override
    public void stop(){}

    @Override
    public void getRevisions(String key, int version, int max, final RevisionsRequestCallbacks callbacks) {
        throw new RuntimeException("Not implemented");
    }

    /**
     * Simulate an acknowledged change
     */
    protected void acknowledge(Change change)
    throws Exception {

        Integer sourceVersion = mBucket.getObject(change.getKey()).getVersion();
        Integer entityVersion;
        if (sourceVersion == null) {
            entityVersion = 1;
        } else {
            entityVersion = sourceVersion + 1;
        }
        JSONArray ccids = new JSONArray();
        String cv = Uuid.uuid(0xF);
        ccids.put(change.getChangeId());

        RemoteChange ack;

        if (!change.getOperation().equals(Change.OPERATION_REMOVE)) {
            Ghost ghost = mBucket.getGhost(change.getKey());
            JSONObject object = mBucket.getObject(change.getKey()).getDiffableValue();
            ack = new RemoteChange("fake", change.getKey(), ccids, cv, sourceVersion, entityVersion, change.getDiff(object, ghost));
        } else {
            ack = new RemoteChange("fake", change.getKey(), ccids, cv, sourceVersion, entityVersion, Change.OPERATION_REMOVE, null);
        }

        Log.d(TAG, String.format("Auto acknowledging %s", ack));
        Log.d(TAG, String.format("Patch: %s", ack.getPatch()));
        ack.isAcknowledgedBy(change);
        mBucket.acknowledgeChange(ack, change);
    }
} public class MockGhostStore implements GhostStorageProvider {
    private Map<String,Map<String,Ghost>> data = new HashMap<String,Map<String,Ghost>>();
    private Map<String,String> versions = new HashMap<String,String>();
    /**
     * Check if the store has a change version for the provided bucket
     */
    @Override
    public boolean hasChangeVersion(Bucket bucket){
        return versions.containsKey(bucket.getName());
    }

    /**
     * Returns wether the bucket's saved version matches the given version
     */
    @Override
    public boolean hasChangeVersion(Bucket bucket, String version){
        return hasChangeVersion(bucket) && versions.get(bucket.getName()).equals(version);
    }

    /**
     * Returns the current change version for the bucket
     */
    @Override
    public String getChangeVersion(Bucket bucket){
        return versions.get(bucket.getName());
    }

    /**
     * Updates the change version for the given bucket
     */
    @Override
    public void setChangeVersion(Bucket bucket, String version){
        versions.put(bucket.getName(), version);
    }

    /**
     * Checks if there is a ghost for given bucket and key
     */

    @Override
    public boolean hasGhost(Bucket bucket, String key){
        return ghostsForBucket(bucket).containsKey(key);
    }

    /**
     * Builds a ghost from the provided bucket and key
     */
    @Override
    public Ghost getGhost(Bucket bucket, String key) throws GhostMissingException {
        Map<String,Ghost> ghosts = ghostsForBucket(bucket);
        Ghost ghost = ghosts.get(key);
        if (ghost == null) {
            throw(new GhostMissingException());
        }
        return ghost;
    }

    @Override
    public int getGhostVersion(Bucket bucket, String key) throws GhostMissingException {
        Map<String,Ghost> ghosts = ghostsForBucket(bucket);
        Ghost ghost = ghosts.get(key);
        if (ghost == null) {
            throw(new GhostMissingException());
        }
        return ghost.getVersion();
    }

    /**
     * Saves the provided ghost to the bucket
     */
    @Override
    public void saveGhost(Bucket bucket, Ghost ghost){
        Map<String,Ghost> ghosts = ghostsForBucket(bucket);
        ghosts.put(ghost.getSimperiumKey(), ghost);
    }

    /**
     * Remove the ghost with the provided key from the store
     */
    @Override
    public void deleteGhost(Bucket bucket, String key){
        Map<String,Ghost> ghosts = ghostsForBucket(bucket);
        ghosts.remove(key);
    }

    /**
     * Clear all ghost data and change version for the given bucket
     */
    @Override
    public void resetBucket(Bucket bucket){
        data.put(bucket.getName(), new HashMap<String,Ghost>());
    }
    
    protected Map<String,Ghost> ghostsForBucket(Bucket bucket){
        String name = bucket.getName();
        Map<String,Ghost> ghosts = data.get(name);
        if (ghosts == null) {
            resetBucket(bucket);
            return ghostsForBucket(bucket);
        }
        return ghosts;
    }
    
} public class MockExecutor {

    public static final String TAG = "Simperium.Test";

    public static class Immediate implements Executor {
        /**
         * Just runs the on the same thread.
         */
        @Override
        public void execute(Runnable runnable){
            runnable.run();
        }

    }

    public static class Playable implements Executor {

        final List<Runnable> pending = new ArrayList<Runnable>();
        boolean mPaused = true;
        String name = "Playable";

        public Playable(){}

        public Playable(String name) {
            this.name = name;
        }

        @Override
        public String toString(){
            String status = mPaused ? "paused" : "playing";
            return String.format(Locale.US, "MockExecutor.Playable %s %s (pending: %d)", name, status, pending.size());
        }

        @Override
        public void execute(Runnable runnable){
            if (!mPaused) {
                runnable.run();
            } else {
                Log.d(TAG, String.format(Locale.US, "%s: Queuing runnable %s", name, runnable));
                pending.add(runnable);
            }
        }

        public void clear(){
            pending.clear();
        }

        public void play(){
            mPaused = false;
            run();
        }

        public void pause(){
            mPaused = true;
        }

        public void run(){
            List<Runnable> tasks = new ArrayList<Runnable>(pending);
            pending.clear();
            Log.d(TAG, String.format("Performing %d tasks", tasks.size()));
            for(Runnable task : tasks){
                Log.d(TAG, String.format("Starting %s", task));
                task.run();
                Log.d(TAG, String.format("Completed %s", task));
            }

            if (pending.size() > 0) {
                Log.d(TAG, String.format("%d Tasks were added, running again", pending.size()));
                run();
            }

        }


    }

    public static MockExecutor.Immediate immediate(){
        return new MockExecutor.Immediate();
    }

} public static User makeUser(String email, String token){
    User user = new User();
    user.setEmail(email);
    user.setAccessToken(token);
    return user;
}
import android.database.sqlite.SQLiteDatabase;
import android.test.ActivityInstrumentationTestCase2;
import com.simperium.client.Bucket;
import com.simperium.client.BucketSchema;
import com.simperium.client.GhostStorageProvider;
import com.simperium.client.User;
import com.simperium.storage.StorageProvider.BucketStore;
import com.simperium.models.Note;
import com.simperium.test.MockChannel;
import com.simperium.test.MockGhostStore;
import com.simperium.test.MockExecutor;
import static com.simperium.TestHelpers.makeUser;
package com.simperium.android;






public abstract class PersistentStoreBaseTest extends ActivityInstrumentationTestCase2<AuthenticationActivity> {

    public static final String MASTER_TABLE = "sqlite_master";
    public static final String BUCKET_NAME="bucket";

    protected AuthenticationActivity mActivity;
    
    protected PersistentStore mStore;
    protected BucketStore<Note> mNoteStore;
    protected SQLiteDatabase mDatabase;
    protected String mDatabaseName = "simperium-test-data";
    protected String[] mTableNames = new String[]{"indexes", "objects", "value_caches"};
    protected Bucket<Note> mBucket;
    protected User mUser;
    protected BucketSchema mSchema;
    protected GhostStorageProvider mGhostStore;

    public PersistentStoreBaseTest() {
        super(AuthenticationActivity.class);
    }

    @Override
    protected void setUp()
    throws Exception {

        super.setUp();

        setActivityInitialTouchMode(false);
        mUser = makeUser();