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 class Change {

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

    public interface OnRetryListener {
        public void onRetry(Change change);
    }

    public interface OnAcknowledgedListener {
        public void onAcknowledged(Change change);
    }

    public interface OnCompleteListener {
        public void onComplete(Change change);
    }

    public static final String OPERATION_MODIFY   = "M";
    public static final String OPERATION_REMOVE   = JSONDiff.OPERATION_REMOVE;
    public static final String ID_KEY             = "id";
    public static final String CHANGE_ID_KEY      = "ccid";
    public static final String SOURCE_VERSION_KEY = "sv";
    public static final String OPERATION_KEY      = "o";
    public static final String OBJECT_DATA_KEY    = "d";

    private String operation;
    private String key, bucketName;
    private String ccid;
    private boolean pending = true, acknowledged = false, sent = false;
    private OnRetryListener retryListener;
    private OnCompleteListener completeListener;
    private OnAcknowledgedListener acknowledgedListener;
    private Change compressed;
    private JSONDiff jsondiff = new JSONDiff();
    private boolean sendFullObject = false;
    private TimerTask retryTimer;
    private Integer retryCount = 0;

    /**
     * Constructs a change object from a map of values
     */
    public static Change buildChange(Syncable object, JSONObject properties)
    throws JSONException {
        return new Change(
            properties.getString(OPERATION_KEY),
            object.getBucketName(),
            object.getSimperiumKey()
        );
    }

    public static Change buildChange(String operation, String ccid, String bucketName, String key){
        return new Change(operation, ccid, bucketName, key);
    }

    public Change(String operation, String bucketName, String key){
        this(operation, uuid(), bucketName, key);
    }

    protected Change(String operation, String ccid, String bucketName, String key){
        this.operation = operation;
        this.ccid = ccid;
        this.bucketName = bucketName;
        this.key = key;

        this.resetTimer();
    }

    public boolean isModifyOperation(){
        return operation.equals(OPERATION_MODIFY);
    }

    public boolean isRemoveOperation() {
        return operation.equals(OPERATION_REMOVE);
    }

    public boolean isPending(){
        return pending;
    }

    public boolean isSent(){
        return sent;
    }

    public boolean isComplete(){
        return !pending;
    }

    public boolean isAcknowledged(){
        return acknowledged;
    }

    protected void setAcknowledged(){
        acknowledged = true;
        resetTimer();
        if (acknowledgedListener != null) {
            acknowledgedListener.onAcknowledged(this);
        }
        if (compressed != null) {
            compressed.setAcknowledged();
        }
    }

    protected void setComplete(){
        pending = false;
        if (completeListener != null) {
            completeListener.onComplete(this);
        }
        if (compressed != null) {
            compressed.setComplete();
        }
    }

    protected void setSent(){
        sent = true;
        if (compressed != null) {
            compressed.setSent();
        }
    }

    protected boolean keyMatches(Change otherChange){
        return otherChange.getKey().equals(getKey());
    }

    public String getKey(){
        return key;
    }

    public String getBucketName(){
        return bucketName;
    }

    public String getChangeId(){
        return this.ccid;
    }

    public String getOperation(){
        return operation;
    }

    public void setSendFullObject(boolean sendFullObject) {
        this.sendFullObject = sendFullObject;
    }

    public JSONObject toJSONObject(JSONObject target, Ghost ghost)
    throws ChangeEmptyException, ChangeInvalidException {
        try {
            JSONObject json = new JSONObject();
            json.put(ID_KEY, getKey());
            json.put(CHANGE_ID_KEY, getChangeId());
            json.put(JSONDiff.DIFF_OPERATION_KEY, getOperation());

            if (!getOperation().equals(OPERATION_REMOVE) && ghost.getVersion() != null && ghost.getVersion() > 0) {
                json.put(SOURCE_VERSION_KEY, ghost.getVersion());
            }

            JSONObject diff = getDiff(target, ghost);
            boolean requiresDiff = requiresDiff();

            if (requiresDiff && diff.length() == 0) {
                throw new ChangeEmptyException(this);
            }

            if (requiresDiff) {
                json.put(JSONDiff.DIFF_VALUE_KEY, diff.getJSONObject(JSONDiff.DIFF_VALUE_KEY));
            }

            if (sendFullObject) {
                json.put(OBJECT_DATA_KEY, target);
            }

            return json;
        } catch (JSONException e) {
            throw new ChangeInvalidException(this, "Could not build change JSON", e);
        }
    }

    protected Map<String,Object> toJSONSerializable(){
        Map<String,Object> props = new HashMap<String,Object>(3);
        // key, version, origin, target, ccid
        props.put(OPERATION_KEY, operation);
        props.put(ID_KEY, key);
        props.put(CHANGE_ID_KEY, ccid);
        return props;

    }

    public void setOnAcknowledgedListener(OnAcknowledgedListener listener){
        acknowledgedListener = listener;
    }

    public void setOnCompleteListener(OnCompleteListener listener){

    }

    public Integer getRetryCount() {
        return retryCount;
    }

    public void incrementRetryCount() {
        retryCount++;
    }

    protected void setOnRetryListener(OnRetryListener listener){
        retryListener = listener;
    }

    protected void resetTimer(){
        if (retryTimer != null) {
            retryTimer.cancel();
        }

        retryTimer = new TimerTask(){
            @Override
            public void run(){
                Logger.log("Simperium.Channel", String.format("Retry change: %s", Change.this));
                if (retryListener != null) {
                    retryListener.onRetry(Change.this);
                }
            }
        };
    }

    protected TimerTask getRetryTimer(){
        return retryTimer;
    }

    public String toString(){
        return String.format("Change %s %s %s", getChangeId(), getKey(), operation);
    }
    /**
     * The change message requires a diff value in the JSON payload
     */
    public boolean requiresDiff(){
        return operation.equals(OPERATION_MODIFY);
    }

    public JSONObject getDiff(JSONObject target, Ghost ghost){
        try {
            return jsondiff.diff(ghost.getDiffableValue(), target);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
} public class Channel implements Bucket.Channel {

    public static class ChangeNotSentException extends ChangeException {

        public ChangeNotSentException(Change change, Throwable cause) {
            super(change, cause);
        }

    }

    public interface OnMessageListener {
        void onMessage(MessageEvent event);
        void onLog(Channel channel, int level, CharSequence message);
        void onClose(Channel channel);
        void onOpen(Channel channel);
    }

    public static final String TAG="Simperium.Channel";
    // key names for init command json object
    static public final String FIELD_CLIENT_ID       = "clientid";
    static public final String FIELD_API_VERSION     = "api";
    static public final String FIELD_AUTH_TOKEN      = "token";
    static public final String FIELD_APP_ID          = "app_id";
    static public final String FIELD_BUCKET_NAME     = "name";
    static public final String FIELD_COMMAND         = "cmd";
    static public final String FIELD_LIBRARY         = "library";
    static public final String FIELD_LIBRARY_VERSION = "version";

    static public final String SIMPERIUM_API_VERSION = "1.1";
    static public final String LIBRARY_NAME = "android";
    static public final Integer LIBRARY_VERSION = 0;
    static public final Integer RETRY_LIMIT = 1;

    // commands sent over the socket
    public static final String COMMAND_INIT      = "init"; // init:{INIT_PROPERTIES}
    public static final String COMMAND_AUTH      = "auth"; // received after an init: auth:expired or auth:email@example.com
    public static final String COMMAND_INDEX     = "i"; // i:1:MARK:?:LIMIT
    public static final String COMMAND_CHANGE    = "c";
    public static final String COMMAND_VERSION   = "cv";
    public static final String COMMAND_ENTITY    = "e";
    static public final String COMMAND_INDEX_STATE = "index";

    static final String RESPONSE_UNKNOWN  = "?";

    static final String EXPIRED_AUTH      = "expired"; // after unsuccessful init:
    static final String EXPIRED_AUTH_INDICATOR = "{";
    static final String EXPIRED_AUTH_CODE_KEY = "code";
    static final int EXPIRED_AUTH_MALFORMED_TOKEN_CODE = 400;
    static final int EXPIRED_AUTH_INVALID_TOKEN_CODE = 401;

    static public final int LOG_DEBUG = ChannelProvider.LOG_DEBUG;

    // Parameters for querying bucket
    static final Integer INDEX_PAGE_SIZE  = 50;

    // Constants for parsing command messages
    static final Integer MESSAGE_PARTS = 2;
    static final Integer COMMAND_PART  = 0;
    static final String  COMMAND_FORMAT = "%s:%s";

    // bucket determines which mBucket we are using on this channel
    private Bucket mBucket;
    // the object the receives the messages the channel emits
    private OnMessageListener mListener;
    // track channel status
    protected boolean mStarted = false, mConnected = false, mStartOnConnect = false, mIdle = true;
    private boolean mHaveIndex = false;
    private CommandInvoker mCommands = new CommandInvoker();
    private String mAppId, mSessionId;
    private Serializer mSerializer;
    protected Executor mExecutor;

    // for sending and receiving changes
    final private ChangeProcessor mChangeProcessor;
    private IndexProcessor mIndexProcessor;
    
    public interface Serializer {
        // public <T extends Syncable> void save(Bucket<T> mBucket, SerializedQueue<T> data);
        SerializedQueue restore(Bucket mBucket);
        void reset(Bucket mBucket);
        void onQueueChange(Change change);
        void onDequeueChange(Change change);
        void onSendChange(Change change);
        void onAcknowledgeChange(Change change);
    }

    public static class SerializedQueue {
        final public Map<String,Change> pending;
        final public List<Change> queued;

        public SerializedQueue() {
            this(new HashMap<String, Change>(), new ArrayList<Change>());
        }

        public SerializedQueue(Map<String,Change> pendingChanges, List<Change> queuedChanges) {
            this.pending = pendingChanges;
            this.queued = queuedChanges;
        }
    }

    public Channel(Executor executor, String appId, String sessionId, final Bucket bucket, Serializer serializer, OnMessageListener listener) {
        mExecutor = executor;
        mSerializer = serializer;
        mAppId = appId;
        mSessionId = sessionId;
        mBucket = bucket;
        mListener = listener;
        // Receive auth: command
        command(COMMAND_AUTH, new Command() {
            public void execute(String param) {
                User user = getUser();
                // ignore auth:expired, implement new auth:{JSON} for failures
                if(EXPIRED_AUTH.equals(param.trim())) return;
                // if it starts with { let's see if it's error JSON
                if (param.indexOf(EXPIRED_AUTH_INDICATOR) == 0) {
                    try {
                        JSONObject authResponse = new JSONObject(param);
                        int code = authResponse.getInt(EXPIRED_AUTH_CODE_KEY);
                        if (code == EXPIRED_AUTH_INVALID_TOKEN_CODE || code == EXPIRED_AUTH_MALFORMED_TOKEN_CODE) {
                            user.setStatus(User.Status.NOT_AUTHORIZED);
                            stop();
                            return;
                        } else {
                            // TODO retry auth?
                            Logger.log(TAG, String.format("Unable to auth: %d", code));
                            return;
                        }
                    } catch (JSONException e) {
                        Logger.log(TAG, String.format("Unable to parse auth JSON, assume was email %s", param));
                    }
                }
                user.setEmail(param);
                user.setStatus(User.Status.AUTHORIZED);
            }
        });

        // Receive i: command
        command(COMMAND_INDEX, new Command() {

            @Override
            public void execute(String param) {
                updateIndex(param);
            }

        });

        // Receive c: command
        command(COMMAND_CHANGE, new Command() {

            @Override
            public void execute(String param) {
                handleRemoteChanges(param);
            }

        });

        // Receive e: command
        command(COMMAND_ENTITY, new Command() {

            @Override
            public void execute(String param) {
                handleVersionResponse(param);
            }

        });

        // Receive index command
        command(COMMAND_INDEX_STATE, new Command() {

            @Override
            public void execute(String param) {
                sendIndexStatus();
            }

        });

        // Receive cv:? command
        command(COMMAND_VERSION, new Command() {

            @Override
            public void execute(String param) {

                if (param.equals(RESPONSE_UNKNOWN)) {
                    Logger.log(TAG, "CV is out of date");
                    stopChangesAndRequestIndex();
                }

            }

        });

        mChangeProcessor = new ChangeProcessor();
    }

    public String toString() {
        return String.format("%s<%s>", super.toString(), mBucket.getName());
    }

    protected Ghost onAcknowledged(RemoteChange remoteChange, Change acknowledgedChange)
    throws RemoteChangeInvalidException {
        // if this isn't a removal, update the ghost for the relevant object
        return mBucket.acknowledgeChange(remoteChange, acknowledgedChange);
    }

    /**
     * Handle errors from simperium service
     * see <a href="https://gist.github.com/beaucollins/6998802#error-responses">Error Responses</a>
     */
    protected void onError(RemoteChange remoteChange, Change erroredChange) {
        RemoteChange.ResponseCode errorCode = remoteChange.getResponseCode();
        switch (errorCode) {
            case INVALID_VERSION:
                // Bad version, client referencing a wrong or missing sv. This is a potential data
                // loss scenario: server may not have enough history to resolve conflicts. Client
                // has two options:
                //
                // - re-load the entity (via e command) then overwrite the local changes
                // - send a change with full object (which will overwrite remote changes, history
                //   will still be available) referencing the current sv
                requeueChangeWithFullObject(erroredChange);
                break;
            case INVALID_DIFF:
                // Server could not apply diff, resend the change with additional parameter d that
                // contains the whole JSON data object. Current known scenarios where this could
                // happen:
                //
                // - Client generated an invalid diff (some old versions of iOS library)
                // - Client is sending string diffs and using a different encoding than server
                requeueChangeWithFullObject(erroredChange);
                break;
            case UNAUTHORIZED:
                // Grab a new authentication token (or possible you just don't
                // have access to that document).

                // TODO: update unauthorized state for User
                break;
            case NOT_FOUND:
                // Client is referencing an object that does not exist on server.
                // If client is insistent, then changing the diff such that it is creating the
                // object instead should make this succeed.
                // TODO: Allow clients to handle NOT_FOUND
                break;
            case INVALID_ID:
                // If it was an invalid id, changing the id could make the call succeed. If it was a
                // schema violation, then correction will depend on the schema. If client cannot
                // tell, then do not re-send since unless something changes, 400 will be sent every
                // time.
                // TODO: Allow client implemenations to handle INVALID_ID
            case EXCEEDS_MAX_SIZE:
                // Nothing to do except reduce the size of the object
                // TODO: Allow client implementations to handle EXCEEDS_MAX_SIZE
            case DUPLICATE_CHANGE:
                // Duplicate change, client can safely throw away the change it is attempting to send
            case EMPTY_CHANGE:
                // Empty change, nothing was changed on the server, client can ignore (and stop
                // sending change).
            case OK:
                // noop
                break;
        }
        Logger.log(TAG, String.format("Received error from service %s", remoteChange));
    }

    @Override
    public void reset() {
        mChangeProcessor.reset();
    }

    private boolean hasChangeVersion() {
        return mBucket.hasChangeVersion();
    }

    private String getChangeVersion() {
        return mBucket.getChangeVersion();
    }

    private void getLatestVersions() {
        mHaveIndex = false;
        // initialize the new query for new index data
        IndexQuery query = new IndexQuery();
        // send the i:::: messages
        sendMessage(query.toString());
    }
    /**
     * Diffs and object's local modifications and queues up the changes
     */
    public Change queueLocalChange(String simperiumKey) {
        Change change = new Change(Change.OPERATION_MODIFY, this.getBucketName(), simperiumKey);
        mChangeProcessor.addChange(change);
        return change;
    }

    public Change queueLocalDeletion(Syncable object) {
        Change change = new Change(Change.OPERATION_REMOVE, this.getBucketName(), object.getSimperiumKey());
        mChangeProcessor.addChange(change);
        return change;
    }

    public void requeueChangeWithFullObject(Change change) {
        // Don't requeue this change if we've retried past the allowed limit.
        if (change.getRetryCount() >= RETRY_LIMIT) {
            completeAndDequeueChange(change);
            return;
        }

        change.incrementRetryCount();
        change.setSendFullObject(true);
        mChangeProcessor.addChange(change);
    }

    private static final String INDEX_CURRENT_VERSION_KEY = "current";
    private static final String INDEX_VERSIONS_KEY = "index";
    private static final String INDEX_MARK_KEY = "mark";

    private void updateIndex(String indexJson) {
        // if we don't have an index processor, create a new one for the associated cv
        // listen for when the index processor is done so we can start the changeprocessor again
        // if we do have an index processor and the cv's match, add the page of items
        // to the queue.
        if (indexJson.equals(RESPONSE_UNKNOWN)) {
            // noop, api 1.1 should not be sending ? here
            return;
        }
        JSONObject index;
        try {
            index = new JSONObject(indexJson);
        } catch (JSONException e) {
            Logger.log(TAG, String.format("Index had invalid json: %s", indexJson));
            return;
        }
        // if we don't have a processor or we are getting a different cv
        if (mIndexProcessor == null || !mIndexProcessor.addIndexPage(index)) {
            // make sure we're not processing changes and clear pending changes
            String currentIndex;
            try {
                currentIndex = index.getString(INDEX_CURRENT_VERSION_KEY);
            } catch(JSONException e) {
                // we have an empty index
                currentIndex = "";
            }

            mIndexProcessor = new IndexProcessor(getBucket(), currentIndex, mIndexProcessorListener);
            mIndexProcessor.start(index);
        } else {
            // received an index page for a different change version
            // TODO: reconcile the out of band index cv
        }
    }

    private IndexProcessorListener mIndexProcessorListener = new IndexProcessorListener() {
        @Override
        public void onComplete(String cv) {
            mHaveIndex = true;
            mIndexProcessor = null;
            mChangeProcessor.start();
        }
    };

    private void handleRemoteChanges(String changesJson) {
        JSONArray changes;
        if (changesJson.equals(RESPONSE_UNKNOWN)) {
            // noop API 1.1 does not send "?" here
            return;
        }
        try {
            changes = new JSONArray(changesJson);
        } catch (JSONException e) {
            Logger.log(TAG, "Failed to parse remote changes JSON", e);
            return;
        }
        // Loop through each change? Convert changes to array list
        mChangeProcessor.addChanges(changes);
    }

    private static final String ENTITY_DATA_KEY = "data";
    private void handleVersionResponse(String versionData) {

        try {
            // versionData will be: key.version\n{"data":ENTITY}
            ObjectVersionData objectVersion = ObjectVersionData.parseString(versionData);

            if (mIndexProcessor != null) {
                mIndexProcessor.addObjectData(objectVersion);
            }

            // if we have any revision requests pending, we want to collect the objects
            boolean collected = false;
            for (RevisionsCollector collector : revisionCollectors) {
                if (collector.addObjectData(objectVersion)) {
                    collected = true;
                }
            }

            if (!collected) {
                updateBucketWithObjectVersion(objectVersion);
            }
        } catch (ObjectVersionUnknownException e) {
            reportRevisionsError();
            log(LOG_DEBUG, String.format(Locale.US, "Object version does not exist %s", e.version));
        } catch (ObjectVersionDataInvalidException e) {
            reportRevisionsError();
            log(LOG_DEBUG, String.format(Locale.US, "Object version JSON data malformed %s", e.version));
        } catch (ObjectVersionParseException e) {
            reportRevisionsError();
            log(LOG_DEBUG, String.format(Locale.US, "Received invalid object version: %s", e.versionString));
        }
    }

    private void updateBucketWithObjectVersion(ObjectVersionData objectVersion) {
        Ghost ghost = new Ghost(objectVersion.getKey(), objectVersion.getVersion(),
            objectVersion.getData());
        mBucket.updateGhost(ghost, null);
    }

    private boolean reportRevisionsError() {
        if (revisionCollectors.size() > 0) {
            for (RevisionsCollector collector : revisionCollectors) {
                // Decrease the expected amount of revisions if we encountered an error
                collector.decreaseRevisionsCount();
            }

            return true;
        }

        return false;
    }

    private void abortRevisionsCollection(Exception e) {
        if (revisionCollectors.size() > 0) {
            for (RevisionsCollector collector : revisionCollectors) {
                if (collector.getCallbacks() != null) {
                    collector.getCallbacks().onError(e);
                }
            }

            revisionCollectors.clear();
        }
    }

    /**
     * Stop sending changes and download a new index
     */
    private void stopChangesAndRequestIndex() {
        // get the latest index
        getLatestVersions();
    }

    /**
     * Send index status JSON
     */
    private void sendIndexStatus() {
        mExecutor.execute(new Runnable() {

            @Override
            public void run() {

                JSONArray objectVersions = new JSONArray();
                String idKey = "id";
                String versionKey = "v";
                Bucket.ObjectCursor objects = mBucket.allObjects();

                // collect all object keys and versions
                while(objects.moveToNext()) {
                    try {
                        JSONObject objectData = new JSONObject();
                        Syncable object = objects.getObject();
                        objectData.put(idKey, object.getSimperiumKey());
                        objectData.put(versionKey, object.getVersion());
                        objectVersions.put(objectData);
                    } catch (JSONException e) {
                        Logger.log(TAG, "Unable to add object version", e);
                    }
                }

                // collect all pending change keys, ccids and source versions
                Collection<Change> pending = mChangeProcessor.pendingChanges();
                JSONArray pendingData = new JSONArray();
                for (Change change : pending) {
                    try {
                        JSONObject changeData = new JSONObject();
                        changeData.put("id", change.getKey());
                        changeData.put("ccid", change.getChangeId());
                        pendingData.put(changeData);
                    } catch (JSONException e) {
                        Logger.log(TAG, "Unable to add change", e);
                    }
                }

                // set up the index information
                JSONObject index = new JSONObject();
                try {
                    index.put("index", objectVersions);
                    index.put("current", getChangeVersion());
                    index.put("pending", pendingData);
                } catch (JSONException e) {
                    Logger.log(TAG, "Unable to build index response", e);
                }

                // add extra debugging info
                JSONObject extra = new JSONObject();
                try {
                    extra.put("bucketName", mBucket.getName());
                    extra.put("build", Version.BUILD);
                    extra.put("version", Version.NUMBER);
                    extra.put("client", Version.NAME);

                    index.put("extra", extra);
                } catch (JSONException e) {
                    Logger.log(TAG, "Unable to add extra info", e);
                }

                sendMessage(String.format("%s:%s", COMMAND_INDEX_STATE, index));
            }

        });
    }

    public Bucket getBucket() {
        return mBucket;
    }

    public String getBucketName() {
        if (mBucket != null) {
            return mBucket.getName();
        }
        return "";
    }

    public User getUser() {
        return mBucket.getUser();
    }

    public String getSessionId() {
        return mSessionId;
    }

    public boolean isStarted() {
        return mStarted;
    }

    public boolean isConnected() {
        return mConnected;
    }

    /**
     * ChangeProcessor has no work to do
     */
    public boolean isIdle() {
        return mIdle;
    }

    @Override
    public void getRevisions(final String key, final int sinceVersion, final int maxVersionCount,
                             final Bucket.RevisionsRequestCallbacks callbacks) {
        // for the key and version iterate down requesting the each version for the object
        RevisionsCollector collector = new RevisionsCollector(key, sinceVersion, maxVersionCount, callbacks);
        revisionCollectors.add(collector);
        collector.send();
    }

    public void log(int level, CharSequence message) {
        if (this.mListener != null) {
            this.mListener.onLog(this, level, message);
        }
    }

    /**
     * Send Bucket's init message to start syncing changes.
     */
    @Override
    public void start() {
        if (mStarted) {
            // we've already mStarted
            return;
        }
        // If socket isn't mConnected yet we have to wait until connection
        // is up and try starting then
        if (!mConnected) {
            if (mListener != null) mListener.onOpen(this);
            mStartOnConnect = true;
            return;
        }
        if (!mBucket.getUser().hasAccessToken()) {
            // we won't connect unless we have an access token
            return;
        }

        mStarted = true;

        Object initialCommand;
        if (!hasChangeVersion()) {
            // the mBucket has never gotten an index
            mHaveIndex = false;
            initialCommand = new IndexQuery();
        } else {
            // retive changes since last cv
            mHaveIndex = true;
            initialCommand = String.format("%s:%s", COMMAND_VERSION, getChangeVersion());
        }

        // Build the required json object for initializing
        HashMap<String,Object> init = new HashMap<>(6);
        init.put(FIELD_API_VERSION, SIMPERIUM_API_VERSION);
        init.put(FIELD_CLIENT_ID, mSessionId);
        init.put(FIELD_APP_ID, mAppId);
        init.put(FIELD_AUTH_TOKEN, mBucket.getUser().getAccessToken());
        init.put(FIELD_BUCKET_NAME, mBucket.getRemoteName());
        init.put(FIELD_COMMAND, initialCommand.toString());
        init.put(FIELD_LIBRARY_VERSION, LIBRARY_VERSION);
        init.put(FIELD_LIBRARY, LIBRARY_NAME);
        String initParams = new JSONObject(init).toString();
        String message = String.format(COMMAND_FORMAT, COMMAND_INIT, initParams);
        sendMessage(message);
    }

    /**
     * Saves syncing state and tells listener to close
     */
    @Override
    public void stop() {
        mStartOnConnect = false;
        mStarted = false;
        if (mListener != null) {
            mListener.onClose(this);
        }
    }

    // websocket
    public void onConnect() {
        mConnected = true;
        Logger.log(TAG, String.format("onConnect autoStart? %b", mStartOnConnect));
        if(mStartOnConnect) start();
    }

    public void onDisconnect() {
        mStarted = false;
        mConnected = false;
    }
    /**
     * Receive a message from the WebSocketManager which already strips the channel
     * prefix from the message.
     */
    public void receiveMessage(String message) {
        // parse the message and react to it
        String[] parts = message.split(":", MESSAGE_PARTS);
        String command = parts[COMMAND_PART];

        if (parts.length == 2) {
            executeCommand(command, parts[1]);
        } else if (parts.length == 1) {
            executeCommand(command, "");
        }

    }

    // send without the channel id, the socket manager should know which channel is writing
    private void sendMessage(String message) {
        // send a message
        if (mListener != null) {
            MessageEvent event = new MessageEvent(this, message);
            mListener.onMessage(event);
        }
    }

    public static class MessageEvent extends EventObject {

        public final String message;
        public final Channel channel;

        public MessageEvent(Channel source, String message) {
            super(source);
            this.message = message;
            this.channel = source;
        }

        public String getMessage() {
            return message;
        }

        public String toString() {
            return getMessage();
        }

        public Channel getChannel() {
            return this.channel;
        }

    }

    private void command(String name, Command command) {
        mCommands.add(name, command);
    }

    private void executeCommand(String name, String params) {
        mCommands.executeCommand(name, params);
    }

    /**
     * Command and CommandInvoker provide a declaritive syntax for handling commands that come in
     * from Channel.onMessage. Takes a message like "auth:user@example.com" and finds the correct
     * command to run and stips the command from the message so the command can take care of
     * processing the params.
     *
     *      channel.command("auth", new Command() {
     *         public void onRun(String params) {
     *           // params is now either an email address or "expired"
     *         }
     *      });
     */
    private interface Command {
        void execute(String params);
    }

    private class CommandInvoker {
        private HashMap<String,Command> mCommands = new HashMap<>();

        protected CommandInvoker add(String name, Command command) {
            mCommands.put(name, command);
            return this;
        }

        protected void executeCommand(String name, String params) {
            if (mCommands.containsKey(name)) {
                Command command = mCommands.get(name);
                command.execute(params);
            } else {
                Logger.log(TAG, String.format("Unkown command received: %s", name));
            }
        }
    }

    static final String CURSOR_FORMAT = "%s::%s::%s";
    /**
     * IndexQuery provides an interface for managing a query cursor and limit fields.
     * TODO: add a way to build an IndexQuery from an index response
     */
    private class IndexQuery {

        private String mMark = "";
        private Integer mLimit = INDEX_PAGE_SIZE;

        public IndexQuery() {}

        public IndexQuery(String mark) {
            this(mark, INDEX_PAGE_SIZE);
        }

        public IndexQuery(String mark, Integer limit) {
            mMark = mark;
            mLimit = limit;
        }

        public String toString() {
            String limitString = "";
            if (mLimit > -1) {
                limitString = mLimit.toString();
            }
            return String.format(CURSOR_FORMAT, COMMAND_INDEX, mMark, limitString);
        }

    }

    public static class ObjectVersion {

        public final String key;
        public final Integer version;

        public ObjectVersion(String key, Integer version) {
            this.key = key;
            this.version = version;
        }

        public String getKey() {
            return key;
        }

        public Integer getVersion() {
            return version;
        }

        public String toString() {
            return String.format(Locale.US, "%s.%d", key, version);
        }

        public static ObjectVersion parseString(String versionString)
        throws ObjectVersionParseException {
            int lastDot = versionString.lastIndexOf(".");
            if (lastDot == -1) {
                throw new ObjectVersionParseException(versionString);
            }
            String key = versionString.substring(0, lastDot);
            String version = versionString.substring(lastDot + 1);
            return new ObjectVersion(key, Integer.parseInt(version));
        }
    }

    public static class ObjectVersionData {

        public final ObjectVersion version;
        public final JSONObject data;

        public ObjectVersionData(ObjectVersion version, JSONObject data) {
            this.version = version;
            this.data = data;
        }

        public String toString() {
            return this.version.toString();
        }

        public String getKey() {
            return this.version.key;
        }

        public Integer getVersion() {
            return this.version.version;
        }

        public JSONObject getData() {
            return this.data;
        }

        public static ObjectVersionData parseString(String versionString)
        throws ObjectVersionParseException, ObjectVersionUnknownException,
        ObjectVersionDataInvalidException {

            String[] objectParts = versionString.split("\n");
            String prefix = objectParts[0];
            String payload = objectParts[1];

            ObjectVersion objectVersion;

            objectVersion = ObjectVersion.parseString(prefix);

            if (payload.equals(RESPONSE_UNKNOWN)) {
                throw new ObjectVersionUnknownException(objectVersion);
            }

            try {
                JSONObject data = new JSONObject(payload);
                return new ObjectVersionData(objectVersion, data.getJSONObject(ENTITY_DATA_KEY));
            } catch (JSONException e) {
                throw new ObjectVersionDataInvalidException(objectVersion, e);
            }

        }

    }

    /**
     * Thrown when e:id.v\n? received
     */
    public static class ObjectVersionUnknownException extends SimperiumException {

        public final ObjectVersion version;

        public ObjectVersionUnknownException(ObjectVersion version) {
            super();
            this.version = version;
        }

    }

    /**
     * Unable to parse the object key and version number from e:key.v
     */
    public static class ObjectVersionParseException extends SimperiumException {

        public final String versionString;

        public ObjectVersionParseException(String versionString) {
            super();
            this.versionString = versionString;
        }

    }

    /**
     * Invalid data received for e:key.v\nJSON
     */
    public static class ObjectVersionDataInvalidException extends SimperiumException {

        public final ObjectVersion version;
        public ObjectVersionDataInvalidException(ObjectVersion version, Throwable cause) {
            super(cause);
            this.version = version;
        }

    }

    // Collects revisions for a Simperium object
    private class RevisionsCollector implements Bucket.RevisionsRequest {

        final private String key;
        final private int sinceVersion;
        final private int maxVersionCount;
        final private Bucket.RevisionsRequestCallbacks callbacks;
        private boolean completed = true;
        private boolean sent = false;
        private int mTotalRevisions;

        private Map<Integer, Syncable> versionsMap = Collections.synchronizedSortedMap(new TreeMap<Integer, Syncable>());

        RevisionsCollector(String key, int sinceVersion, int maxVersionCount, Bucket.RevisionsRequestCallbacks callbacks) {
            this.key = key;
            this.sinceVersion = sinceVersion;
            this.maxVersionCount = maxVersionCount;
            this.callbacks = callbacks;
        }

        private void send() {
            if (!mConnected) {
                if (callbacks != null) {
                    abortRevisionsCollection(new Exception("Can't retrieve revisions: No connection."));
                }

                return;
            }

            if (!sent) {
                sent = true;
                int minVersion;
                if (maxVersionCount > 0) {
                    minVersion = (sinceVersion - maxVersionCount > 0) ? sinceVersion - maxVersionCount : 1;
                } else {
                    minVersion = 1;
                }
                mTotalRevisions = sinceVersion - minVersion;
                // for each version send an e: request
                for (int i = minVersion; i < sinceVersion; i++) {
                    sendObjectVersionRequest(key, i);
                }
            }
        }

        public boolean addObjectData(ObjectVersionData objectVersionData) {
            int version = objectVersionData.getVersion();
            if (objectVersionData.getKey().equals(this.key) && version < sinceVersion && versionsMap.get(version) == null) {
                versionsMap.put(version, mBucket.buildObject(this.key, objectVersionData.getData()));

                JSONObject data = objectVersionData.getData();
                callbacks.onRevision(key, version, data);

                if (versionsMap.size() == mTotalRevisions) {
                    revisionCollectors.remove(this);
                    completed = true;
                    callbacks.onComplete(versionsMap);
                }
                return true;
            }
            return false;
        }

        public Bucket.RevisionsRequestCallbacks getCallbacks() {
            return callbacks;
        }

        @Override
        public boolean isComplete() {
            return completed;
        }

        public void decreaseRevisionsCount() {
            mTotalRevisions--;
        }
    }

    private void sendObjectVersionRequest(String key, int version){
        sendMessage(String.format("%s:%s.%d", COMMAND_ENTITY, key, version));
    }

    private List<RevisionsCollector> revisionCollectors = Collections.synchronizedList(new ArrayList<RevisionsCollector>());

    private interface IndexProcessorListener {
        
        void onComplete(String cv);
    }

    /**
     * When index data is received it should queue up entities in the IndexProcessor.
     * The IndexProcessor then receives the object data and on a seperate thread asks
     * the StorageProvider to persist the object data. The storageProvider's operation
     * should not block the websocket thread in any way.
     *
     * Build up a list of entities and versions we need for the index. Allow the
     * channel to pass in the version data
     */
    private class IndexProcessor {

        public static final String INDEX_OBJECT_ID_KEY = "id";
        public static final String INDEX_OBJECT_VERSION_KEY = "v";

        final private String mCv;
        final private Bucket mBucket;
        private List<String> mQueue = Collections.synchronizedList(new ArrayList<String>());
        private IndexQuery mNextQuery;
        protected boolean mComplete = false, mNotified = false;
        final private IndexProcessorListener mListener;
        protected int mIndexedCount = 0, mReceivedCount = 0;
        final protected Object mCountLock;

        public IndexProcessor(Bucket bucket, String cv, IndexProcessorListener listener) {
            mBucket = bucket;
            mCv = cv;
            mListener = listener;
            mCountLock = new Object();
        }

        /**
         * Receive an object's version data and store it. Send the request for the
         * next object.
         */
        public boolean addObjectData(ObjectVersionData objectVersion) {
            if (!mQueue.remove(objectVersion.toString())) {
                return false;
            }

            // build the ghost and update
            Ghost ghost = new Ghost(objectVersion.getKey(), objectVersion.getVersion(), objectVersion.getData());
            mBucket.addObjectWithGhost(ghost, new Runnable() {
                @Override
                public void run() {
                    synchronized(mCountLock) {
                        mIndexedCount ++;
                        if (mIndexedCount % 10 == 0) {
                            notifyProgress();
                        }
                    }

                    if (mComplete && mReceivedCount == mIndexedCount) {
                        notifyDone();
                    }
                }

            });

            next();

            return true;
        }

        public void start(JSONObject indexPage) {
            addIndexPage(indexPage);
        }

        public void next() {

            // if queue isn't empty, pull it off the top and send request
            if (!mQueue.isEmpty()) {
                String versionString = mQueue.get(0);
                ObjectVersion version;
                try {
                    version = ObjectVersion.parseString(versionString);
                } catch (ObjectVersionParseException e) {
                    Logger.log(TAG, "Failed to parse version string, skipping", e);
                    mQueue.remove(versionString);
                    next();
                    return;
                }

                if (!mBucket.hasKeyVersion(version.getKey(), version.getVersion())) {
                    sendMessage(String.format("%s:%s", COMMAND_ENTITY, version.toString()));
                } else {
                    synchronized(mCountLock) {
                        mIndexedCount ++;
                        Logger.log(TAG, String.format("Already have %s requesting next object", version));
                        mQueue.remove(versionString);
                    }
                    next();
                    return;
                }
                return;
            }

            // if index is empty and we have a next request, make the request
            if (mNextQuery != null) {
                sendMessage(mNextQuery.toString());
                return;
            }

            // no queue, no next query, all done!
            mComplete = true;
            if (BuildConfig.DEBUG) {
                Log.d(TAG, "Done receiving object data " + Channel.this);
            }

            if (mReceivedCount == mIndexedCount) {
                notifyDone();
            }

        }

        /**
         * Add the page of data, but only if indexPage cv matches. Detects when it's the
         * last page due to absence of cursor mark
         */
        public Boolean addIndexPage(JSONObject indexPage) {

            String currentIndex;
            try {
                currentIndex = indexPage.getString(INDEX_CURRENT_VERSION_KEY);
            } catch(JSONException e) {
                Logger.log(TAG, String.format("Index did not have current version %s", mCv));
                currentIndex = "";
            }

            if (!currentIndex.equals(mCv)) {
                return false;
            }

            JSONArray indexVersions;
            try {
                indexVersions = indexPage.getJSONArray(INDEX_VERSIONS_KEY);
            } catch(JSONException e) {
                Logger.log(TAG, String.format("Index did not have entities: %s", indexPage));
                return true;
            }

            if (indexVersions.length() > 0) {
                // query for each item that we don't have locally in the mBucket
                for (int i=0; i<indexVersions.length(); i++) {
                    try {

                        JSONObject version = indexVersions.getJSONObject(i);
                        String key  = version.getString(INDEX_OBJECT_ID_KEY);
                        Integer versionNumber = version.getInt(INDEX_OBJECT_VERSION_KEY);
                        ObjectVersion objectVersion = new ObjectVersion(key, versionNumber);
                        mQueue.add(objectVersion.toString());

                        synchronized(mCountLock) {
                            mReceivedCount ++;
                        }

                    } catch (JSONException e) {
                        Logger.log(TAG, String.format("Error processing index: %d", i), e);
                    }

                }

            }

            String nextMark = null;
            if (indexPage.has(INDEX_MARK_KEY)) {
                try {
                    nextMark = indexPage.getString(INDEX_MARK_KEY);
                } catch (JSONException e) {
                    nextMark = null;
                }
            }

            if (nextMark != null && nextMark.length() > 0) {
                mNextQuery = new IndexQuery(nextMark);
            } else {
                mNextQuery = null;
            }
            next();
            return true;
        }

        synchronized private void notifyDone() {
            if (mNotified) {
                return;
            }
            mNotified = true;
            if (BuildConfig.DEBUG) {
                Log.d(TAG, "Notifying index done: " + Channel.this);
            }
            mBucket.indexComplete(mCv);
            mListener.onComplete(mCv);
        }


        private void notifyProgress() {
            if (BuildConfig.DEBUG) {
                Log.d(TAG, "Notifying index progress: " + Channel.this);
            }
            mBucket.notifyOnNetworkChangeListeners(Bucket.ChangeType.INDEX);
        }
    }

    public boolean haveCompleteIndex() {
        return mHaveIndex;
    }

    /**
     * ChangeProcessor should perform operations on a seperate thread as to not block the websocket
     * ideally it will be a FIFO queue processor so as changes are brought in they can be appended.
     * We also need a way to pause and clear the queue when we download a new index.
     */
    private class ChangeProcessor implements Runnable, Change.OnRetryListener {

        // wait 5 seconds for retries
        public static final long RETRY_DELAY_MS = 5000; 
        private List<JSONObject> mRemoteQueue = Collections.synchronizedList(new ArrayList<JSONObject>(10));
        private List<Change> mLocalQueue = Collections.synchronizedList(new ArrayList<Change>());
        private Map<String,Change> mPendingChanges = Collections.synchronizedMap(new HashMap<String,Change>());
        private Timer mRetryTimer;
        private Thread mThread;
        private final Object mLock = new Object();
        private final Object mRunLock = new Object();

        public ChangeProcessor() {
            restore();
        }

        public Collection<Change> pendingChanges() {
            return mPendingChanges.values();
        }

        private void restore() {
            synchronized(mLock) {
                SerializedQueue serialized = mSerializer.restore(mBucket);
                mLocalQueue.addAll(serialized.queued);
                mPendingChanges.putAll(serialized.pending);
                resendPendingChanges();
            }
        }

        public void addChanges(JSONArray changes) {
            synchronized(mLock) {
                int length = changes.length();
                Logger.log(TAG, String.format("Add remote changes to processor %d", length));
                log(LOG_DEBUG, String.format(Locale.US, "Adding %d remote changes to queue", length));
                for (int i = 0; i < length; i++) {
                    JSONObject change = changes.optJSONObject(i);
                    if (change != null) {
                        mRemoteQueue.add(change);
                    }
                }
                start();
            }
        }

        /**
         * Local change to be queued
         */
        public void addChange(Change change) {
            synchronized(mLock) {
                // compress all changes for this same key
                log(LOG_DEBUG, String.format(Locale.US, "Adding new change to queue %s %s %s",
                    change.getKey(), change.getOperation(), change.getChangeId()));

                Iterator<Change> iterator = mLocalQueue.iterator();
                boolean isModify = change.isModifyOperation();
                while(iterator.hasNext() && isModify) {
                    Change queued = iterator.next();
                    if(queued.getKey().equals(change.getKey())) {
                        mSerializer.onDequeueChange(queued);
                        iterator.remove();
                    }
                }
                mSerializer.onQueueChange(change);
                mLocalQueue.add(change);
                notifyLocalQueueChange();
            }
            start();
        }

        public void start() {
            // channel must be started and have complete index
            if (!mStarted) {
                return;
            }
            if (mRetryTimer == null) {
                mRetryTimer = new Timer();
            }
            if (mThread == null || mThread.getState() == Thread.State.TERMINATED) {
                mThread = new Thread(this, String.format("simperium.processor.%s", getBucket().getName()));
                mThread.start();
            } else {
                // notify
                synchronized(mRunLock) {
                    mRunLock.notify();
                }
            }
        }

        protected void reset() {
            mPendingChanges.clear();
            mSerializer.reset(mBucket);
        }

        protected void abort() {
            reset();
        }

        /**
         * Check if we have changes we can send out
         */
        protected boolean hasQueuedChanges() {
            synchronized(mLock) {
                Logger.log(TAG, String.format("Checking for queued changes %d", mLocalQueue.size()));
                // if we have have any remote changes to process we have work to do
                if (!mRemoteQueue.isEmpty()) return true;
                // if our local queue is empty we don't have work to do
                if (mLocalQueue.isEmpty()) return false;
                // if we have queued changes, if there's no corresponding pending change then there's still work to do
                for (Change change : mLocalQueue) {
                    if (!mPendingChanges.containsKey(change.getKey())) return true;
                }
                return false;
            }
        }

        protected void notifyLocalQueueChange() {
            // send list of entity ids which still have unconfirmed local changes
            Set<String> queuedKeys = new HashSet<>();
            for (Change change : mLocalQueue) {
                queuedKeys.add(change.getKey());
            }
            queuedKeys.addAll(mPendingChanges.keySet());

            mBucket.notifyOnLocalQueueChangeListeners(queuedKeys);
        }

        public void run() {
            if(!haveCompleteIndex()) return;
            mIdle = false;
            Logger.log(TAG, String.format("%s - Starting change queue", Thread.currentThread().getName()));
            while(true) {
                try {
                    processRemoteChanges();
                    processLocalChanges();
                } catch (InterruptedException e) {
                    // shut down
                    break;
                }
                if(!hasQueuedChanges()) {
                    // we've sent out every change that we can so far, if nothing is pending we can disconnect
                    if (mPendingChanges.isEmpty()) {
                        mIdle = true;
                    }

                    synchronized(mRunLock) {
                        try {
                            Logger.log(TAG, String.format("Waiting <%s> mIdle? %b", mBucket.getName(), mIdle));
                            log(LOG_DEBUG, "Change queue is empty, waiting for changes");
                            mRunLock.wait();
                        } catch (InterruptedException e) {
                            break;
                        }
                        Logger.log(TAG, "Waking change processor");
                        log(LOG_DEBUG, "Processing changes");
                    }
                }
            }
            mRetryTimer.cancel();
            mRetryTimer = null;
            Logger.log(TAG, String.format("%s - Queue interrupted", Thread.currentThread().getName()));
        }


        private void processRemoteChanges()
        throws InterruptedException {
            synchronized(mLock) {
                Logger.log(TAG, String.format("Processing remote changes %d", mRemoteQueue.size()));
                // bail if thread is interrupted
                while(mRemoteQueue.size() > 0) {
                    if (Thread.interrupted()) {
                        throw new InterruptedException();
                    }
                    // take an item off the queue
                    RemoteChange remoteChange;
                    try {
                        remoteChange = RemoteChange.buildFromMap(mRemoteQueue.remove(0));
                    } catch (JSONException e) {
                        Logger.log(TAG, "Failed to build remote change", e);
                        continue;
                    }
                    log(LOG_DEBUG, String.format("Processing remote change with cv: %s", remoteChange.getChangeVersion()));
                    // synchronizing on pendingChanges since we're looking up and potentially
                    // removing an entry
                    Change change = mPendingChanges.get(remoteChange.getKey());
                    if (remoteChange.isAcknowledgedBy(change)) {
                        log(LOG_DEBUG, String.format("Found pending change for remote change <%s>: %s", remoteChange.getChangeVersion(), change.getChangeId()));
                        mSerializer.onAcknowledgeChange(change);
                        // change is no longer pending so remove it
                        mPendingChanges.remove(change.getKey());
                        notifyLocalQueueChange();
                        if (remoteChange.isError()) {
                            Logger.log(TAG, String.format("Change error response! %d %s", remoteChange.getErrorCode(), remoteChange.getKey()));
                            onError(remoteChange, change);
                        } else {
                            Ghost ghost = null;
                            try {
                                ghost = onAcknowledged(remoteChange, change);
                            } catch (RemoteChangeInvalidException e) {
                                Logger.log(TAG, "Remote change could not be acknowledged", e);
                                log(LOG_DEBUG, String.format("Failed to acknowledge change <%s> Reason: %s", remoteChange.getChangeVersion(), e.getMessage()));

                                // request the full object for the new version
                                ObjectVersion version = new ObjectVersion(remoteChange.getKey(), remoteChange.getObjectVersion());
                                sendMessage(String.format("%s:%s", COMMAND_ENTITY, version));
                            } finally {
                                Change compressed = null;
                                Iterator<Change> queuedChanges = mLocalQueue.iterator();
                                while (queuedChanges.hasNext()) {
                                    Change queuedChange = queuedChanges.next();
                                    if (queuedChange.getKey().equals(change.getKey())) {
                                        queuedChanges.remove();
                                        if (ghost != null && !remoteChange.isRemoveOperation()) {
                                            compressed = queuedChange;
                                        }
                                    }
                                }
                                if (compressed != null) {
                                    mLocalQueue.add(compressed);
                                }
                            }
                        }
                    } else {
                        if (remoteChange.isError()) {
                            Logger.log(TAG, String.format("Remote change %s was an error but not acknowledged", remoteChange));
                            log(LOG_DEBUG, String.format("Received error response for change but not waiting for any ccids <%s>", remoteChange.getChangeVersion()));
                        } else {
                            try {
                                mBucket.applyRemoteChange(remoteChange);
                                Logger.log(TAG, String.format("Succesfully applied remote change <%s>", remoteChange.getChangeVersion()));
                            } catch (RemoteChangeInvalidException e) {
                                Logger.log(TAG, "Remote change could not be applied", e);
                                log(LOG_DEBUG, String.format("Failed to apply change <%s> Reason: %s", remoteChange.getChangeVersion(), e.getMessage()));

                                dequeueLocalChangesForKey(remoteChange.getKey());
                                // request the full object for the new version
                                ObjectVersion version = new ObjectVersion(remoteChange.getKey(), remoteChange.getObjectVersion());
                                sendMessage(String.format("%s:%s", COMMAND_ENTITY, version));
                            }
                        }
                    }

                    boolean wasGoodChange = !remoteChange.isError();
                    boolean entityIsGone = wasGoodChange && remoteChange.isRemoveOperation();

                    if (entityIsGone) {
                        dequeueLocalChangesForKey(remoteChange.getKey());
                    }

                    if (wasGoodChange && !entityIsGone) {
                        mBucket.notifyOnSyncObjectListeners(remoteChange.getKey());
                    }
                }
            }
        }

        private void dequeueLocalChangesForKey(String simperiumKey) {
            if (simperiumKey == null) return;

            Iterator<Change> iterator = mLocalQueue.iterator();
            while (iterator.hasNext()) {
                Change queuedChange = iterator.next();
                if (queuedChange.getKey().equals(simperiumKey)) {
                    iterator.remove();
                }
            }
        }

        public void processLocalChanges()
        throws InterruptedException {
            synchronized(mLock) {
                if (mLocalQueue.isEmpty()) {
                    return;
                }
                final List<Change> sendLater = new ArrayList<>();
                // find the first local change whose key does not exist in the pendingChanges and there are no remote changes
                while(!mLocalQueue.isEmpty()) {
                    if (Thread.interrupted()) {
                        mLocalQueue.addAll(0, sendLater);
                        throw new InterruptedException();
                    }

                    // take the first change of the queue
                    Change localChange = mLocalQueue.remove(0);
                        // check if there's a pending change with the same key
                    if (mPendingChanges.containsKey(localChange.getKey())) {
                        // we have a change for this key that has not been acked
                        // so send it later
                        sendLater.add(localChange);
                        // let's get the next change
                    } else {
                        try {
                            // add the change to pending changes
                            mPendingChanges.put(localChange.getKey(), localChange);
                            // send the change to simperium, if the change ends up being empty
                            // then we'll just skip it
                            sendChange(localChange);
                            localChange.setOnRetryListener(this);
                            // starts up the timer
                            mRetryTimer.scheduleAtFixedRate(localChange.getRetryTimer(), RETRY_DELAY_MS, RETRY_DELAY_MS);
                        } catch (ChangeNotSentException e) {
                            mPendingChanges.remove(localChange.getKey());
                            notifyLocalQueueChange();
                        }
                    }
                }
                mLocalQueue.addAll(0, sendLater);
            }
        }

        private void resendPendingChanges() {
            if (mRetryTimer == null) {
                mRetryTimer = new Timer();
            }
            synchronized(mLock) {
                // resend all pending changes
                for (Map.Entry<String, Change> entry : mPendingChanges.entrySet()) {
                    Change change = entry.getValue();
                    change.setOnRetryListener(this);
                    mRetryTimer.scheduleAtFixedRate(change.getRetryTimer(), RETRY_DELAY_MS, RETRY_DELAY_MS);
                }
            }
        }

        @Override
        public void onRetry(Change change) {
            log(LOG_DEBUG, String.format("Retrying change %s", change.getChangeId()));
            try {
                sendChange(change);
            } catch (ChangeNotSentException e) {
                // do nothing the timer will trigger another send
            }
        }

        private void sendChange(Change change)
        throws ChangeNotSentException {
            // send the change down the socket!
            if (!mConnected) {
                // channel is not initialized, send on reconnect
                return;
            }

            try {
                log(LOG_DEBUG, String.format("Sending change for id: %s op: %s ccid: %s", change.getKey(), change.getOperation(), change.getChangeId()));
                Syncable target = mBucket.getObjectOrBackup(change.getKey());
                Ghost ghost = mBucket.getGhost(change.getKey());
                sendMessage(String.format("c:%s", change.toJSONObject(target.getDiffableValue(), ghost)));
                mSerializer.onSendChange(change);
                change.setSent();
            } catch (BucketObjectMissingException e) {
                Logger.log("Could not get object to send change");
                completeAndDequeueChange(change);
                throw new ChangeNotSentException(change, e);
            } catch (GhostMissingException e) {
                Logger.log("Could not get ghost to send change");
                completeAndDequeueChange(change);
                throw new ChangeNotSentException(change, e);
            } catch (ChangeEmptyException e) {
                completeAndDequeueChange(change);
                throw new ChangeNotSentException(change, e);
            } catch (ChangeException e) {
                android.util.Log.e(TAG, "Could not send change", e);
                throw new ChangeNotSentException(change, e);
            }

        }

    }

    private void completeAndDequeueChange(Change change) {
        change.setComplete();
        change.resetTimer();
        mSerializer.onDequeueChange(change);
    }

    public static Map<String,Object> convertJSON(JSONObject json) {
        Map<String,Object> map = new HashMap<String,Object>(json.length());
        Iterator keys = json.keys();
        while(keys.hasNext()) {
            String key = (String)keys.next();
            try {
                Object val = json.get(key);
                if (val.getClass().equals(JSONObject.class)) {
                    map.put(key, convertJSON((JSONObject) val));
                } else if (val.getClass().equals(JSONArray.class)) {
                    map.put(key, convertJSON((JSONArray) val));
                } else {
                    map.put(key, val);
                }
            } catch (JSONException e) {
                Logger.log(TAG, String.format("Failed to convert JSON: %s", e.getMessage()), e);
            }
        }
        return map;
    }

    public static List<Object> convertJSON(JSONArray json) {
        List<Object> list = new ArrayList<Object>(json.length());
        for (int i=0; i<json.length(); i++) {
            try {
                Object val = json.get(i);
                if (val.getClass().equals(JSONObject.class)) {
                    list.add(convertJSON((JSONObject) val));
                } else if (val.getClass().equals(JSONArray.class)) {
                    list.add(convertJSON((JSONArray) val));
                } else {
                    list.add(val);
                }
            } catch (JSONException e) {
                Logger.log(TAG, String.format("Failed to convert JSON: %s", e.getMessage()), e);
            }

        }
        return list;
    }

    public static JSONObject serializeJSON(Map<String,Object>map) {
        JSONObject json = new JSONObject();
        for (String key : map.keySet()) {
            Object val = map.get(key);
            try {
                if (val instanceof Map) {
                    json.put(key, serializeJSON((Map<String, Object>) val));
                } else if (val instanceof List) {
                    json.put(key, serializeJSON((List<Object>) val));
                } else if (val instanceof Change) {
                    json.put(key, serializeJSON(((Change) val).toJSONSerializable()));
                } else {
                    json.put(key, val);
                }
            } catch (JSONException e) {
                Logger.log(TAG, String.format("Failed to serialize %s", val));
            }
        }
        return json;
    }

    public static JSONArray serializeJSON(List<Object>list) {
        JSONArray json = new JSONArray();
        for (Object val : list) {
            if (val instanceof Map) {
                json.put(serializeJSON((Map<String, Object>) val));
            } else if (val instanceof List) {
                json.put(serializeJSON((List<Object>) val));
            } else if (val instanceof Change) {
                json.put(serializeJSON(((Change) val).toJSONSerializable()));
            } else {
                json.put(val);
            }
        }
        return json;
    }

} public interface ChannelProvider {

    static public final int LOG_DISABLED = 0;
    static public final int LOG_DEBUG = 1;
    static public final int LOG_VERBOSE = 2;

    public Bucket.Channel buildChannel(Bucket bucket);

    /**
     * Send a log message to Simperium
     */
    public void log(int level, CharSequence message);


    /**
     * Get the current log level set from the service
     */
    public int getLogLevel();


    public interface HeartbeatListener {
        public void onBeat();
    }

    public void addHeartbeatListener(HeartbeatListener listener);
} public class RemoteChange {

    /**
     * Possible simperium response errors
     * @see https://gist.github.com/beaucollins/6998802#error-responses
     */
    public enum ResponseCode {
        OK               (200),
        INVALID_ID       (400),
        UNAUTHORIZED     (401),
        NOT_FOUND        (404),
        INVALID_VERSION  (405),
        DUPLICATE_CHANGE (409),
        EMPTY_CHANGE     (412),
        EXCEEDS_MAX_SIZE (413),
        INVALID_DIFF     (440);

        public final int code;

        ResponseCode(int code) {
            this.code = code;
        }

    }

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

    public static final String ID_KEY             = "id";
    public static final String CLIENT_KEY         = "clientid";
    public static final String ERROR_KEY          = "error";
    public static final String END_VERSION_KEY    = "ev";
    public static final String SOURCE_VERSION_KEY = "sv";
    public static final String CHANGE_VERSION_KEY = "cv";
    public static final String CHANGE_IDS_KEY     = "ccids";
    public static final String OPERATION_KEY      = JSONDiff.DIFF_OPERATION_KEY;
    public static final String VALUE_KEY          = JSONDiff.DIFF_VALUE_KEY;
    public static final String OPERATION_MODIFY   = "M";
    public static final String OPERATION_REMOVE   = JSONDiff.OPERATION_REMOVE;

    private String key;
    private String clientid;
    private JSONArray ccids;
    private Integer sourceVersion;
    private Integer entityVersion;
    private String changeVersion;
    private String operation;
    private JSONObject value;
    private Integer errorCode;
    private boolean applied = false;
    private Change change;
    private JSONDiff jsondiff = new JSONDiff();

    /**
     * All remote changes include clientid, key and ccids then these differences:
     * - errors have error key
     * - changes with operation "-" do not have a value
     * - changes with operation "M" have a v (value), ev (entity version) and if not a new object an sv (source version)
     */
    public RemoteChange(String clientid, String key, JSONArray ccids, String changeVersion,
        Integer sourceVersion, Integer entityVersion, String operation, JSONObject value) {
        this.clientid = clientid;
        this.key = key;
        this.ccids = ccids;
        this.sourceVersion = sourceVersion;
        this.entityVersion = entityVersion;
        this.operation = operation;
        this.value = value;
        this.changeVersion = changeVersion;
    }

    public RemoteChange(String clientid, String key, JSONArray ccids, String changeVersion,
        Integer sourceVersion, Integer entityVersion, JSONObject diff)
        throws JSONException {
        this(clientid, key, ccids, changeVersion, sourceVersion, entityVersion,
            diff.getString(JSONDiff.DIFF_OPERATION_KEY),
            diff.getJSONObject(JSONDiff.DIFF_VALUE_KEY));
    }

    public RemoteChange(String clientid, String key, JSONArray ccids, Integer errorCode){
        this.clientid = clientid;
        this.key = key;
        this.ccids = ccids;
        this.errorCode = errorCode;
    }

    protected Ghost apply(Ghost ghost) throws RemoteChangeInvalidException {
        // keys and versions must match otherwise throw an error
        if (!ghost.getSimperiumKey().equals(getKey())) {
            throw(new RemoteChangeInvalidException(this,
                    String.format(Locale.US, "Local instance key %s does not match change key %s",
                        ghost.getSimperiumKey(), getKey())));
        }

        if (isModifyOperation() && !ghost.getVersion().equals(getSourceVersion())) {
            throw(new RemoteChangeInvalidException(this,
                    String.format(Locale.US, "Local instance of %s has source version of %d and remote change has %d",
                    getKey(), ghost.getVersion(), getSourceVersion())));
        }

        if (isAddOperation() && ghost.getVersion() != 0) {
            throw(new RemoteChangeInvalidException(this, "Local instance has version greater than 0 with remote add operation"));
        }

        try {
            JSONObject properties = jsondiff.apply(ghost.getDiffableValue(), getPatch());
            return new Ghost(getKey(), getObjectVersion(), properties);
        } catch (JSONException e) {
            throw new RemoteChangeInvalidException(this, String.format("Unable to apply patch: %s", getPatch()), e);
        } catch (IllegalArgumentException e) {
            throw new RemoteChangeInvalidException(this, "Invalid patch", e);
        }

    }

    public boolean isAcknowledged(){
        return change != null;
    }

    public boolean isApplied(){
        return applied;
    }

    protected void setApplied(){
        if (applied == false) {
            applied = true;
            if(change != null) change.setComplete();
        }
    }

    public boolean isAcknowledgedBy(Change change){
        if (change == null) return false;
        // if we have a Change with the same change id from the same client id
        // then we were waiting for this change
        if(hasChangeId(change)){
            this.change = change;
            change.setAcknowledged();
            return true;
        }
        return false;
    }

    public boolean isError(){
        return errorCode != null;
    }

    public String getOperation(){
        return operation;
    }

    public boolean isRemoveOperation(){
        if (operation == null) return false;
        return operation.equals(OPERATION_REMOVE);
    }

    public boolean isModifyOperation(){
        if (operation == null) return false;
        return operation.equals(OPERATION_MODIFY) && sourceVersion != null && sourceVersion > 0;
    }

    public boolean isAddOperation(){
        if (operation == null) return false;
        return operation.equals(OPERATION_MODIFY) && (sourceVersion == null || sourceVersion <= 0);
    }

    public ResponseCode getResponseCode() {
        return responseForCode(getErrorCode());
    }

    public Integer getErrorCode(){
        return errorCode;
    }

    public boolean isNew(){
        return !isError() && (!hasSourceVersion() || sourceVersion.equals(0));
    }

    public String getKey(){
        return key;
    }

    public String getClientId(){
        return clientid;
    }

    public boolean hasSourceVersion(){
        return sourceVersion != null;
    }

    public Integer getSourceVersion(){
        if (sourceVersion == null) {
            return 0;
        }
        return sourceVersion;
    }

    public Integer getObjectVersion(){
        return entityVersion;
    }

    public String getChangeVersion(){
        return changeVersion;
    }

    public JSONObject getPatch(){
        return value;
    }

    public boolean hasChangeId(String ccid){

        if (ccid == null) return false;

        int length = ccids.length();
        for (int i=0; i<length; i++) {
            try {
                if (ccid.equals(ccids.getString(i)))
                    return true;;
            } catch (JSONException e) {
                android.util.Log.e(TAG, "Invalid ccid", e);
            }
        }

        return false;
    }

    public boolean hasChangeId(Change change){
        return hasChangeId(change.getChangeId());
    }

    public JSONArray getChangeIds(){
        return ccids;
    }

    public String toString(){
        if (!isError()) {
            return String.format(Locale.US, "RemoteChange %s %s %s %d-%d", getKey(), getChangeVersion(), operation, getSourceVersion(), getObjectVersion());
        } else {
            return String.format(Locale.US, "RemoteChange %s Error %d", getKey(), getErrorCode());
        }
    }

    public static RemoteChange buildFromMap(JSONObject changeData)
    throws JSONException {
        // get the list of ccids that this applies to
        JSONArray ccids = changeData.getJSONArray(CHANGE_IDS_KEY);
        // get the client id
        String client_id = changeData.getString(CLIENT_KEY);
        // get the id of the object it applies to
        String id = changeData.getString(ID_KEY);
        if (changeData.has(ERROR_KEY)) {
            int errorCode = changeData.getInt(ERROR_KEY);
            return new RemoteChange(client_id, id, ccids, errorCode);
        }

        String operation = changeData.getString(OPERATION_KEY);
        Integer sourceVersion = changeData.optInt(SOURCE_VERSION_KEY);
        Integer objectVersion = changeData.optInt(END_VERSION_KEY);
        JSONObject patch = changeData.optJSONObject(VALUE_KEY);
        String changeVersion = changeData.getString(CHANGE_VERSION_KEY);

        return new RemoteChange(client_id, id, ccids, changeVersion, sourceVersion, objectVersion, operation, patch);

    }

    public static ResponseCode responseForCode(Integer code) {

        if (code == null)
            return ResponseCode.OK;

        for (ResponseCode responseCode : ResponseCode.values()) {
            if (responseCode.code == code) {
                return responseCode;
            }
        }

        return ResponseCode.OK;
    }

} 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 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 MockBucket<T extends Syncable> extends Bucket<T> {

    protected RemoteChangeListener mListener;
    public MockGhostStore ghostStore;

    public interface RemoteChangeListener {
        public void onApplyRemoteChange(RemoteChange change);
        public void onAcknowledgeRemoteChange(RemoteChange change);
    }

    public MockBucket(Executor executor, String name, BucketSchema<T>schema, User user, BucketStore<T> storage, MockGhostStore ghostStore)
    throws BucketNameInvalid {
        super(executor, name, schema, user, storage, ghostStore);
        this.ghostStore = ghostStore;
    }

    public void setRemoteChangeListener(RemoteChangeListener listener){
        mListener = listener;
    }

    @Override
    public Ghost applyRemoteChange(RemoteChange change)
    throws RemoteChangeInvalidException {
        if (mListener != null) mListener.onApplyRemoteChange(change);
        return super.applyRemoteChange(change);
    }

    @Override
    public Ghost acknowledgeChange(RemoteChange remoteChange, Change change)
    throws RemoteChangeInvalidException {
        if (mListener != null) mListener.onAcknowledgeRemoteChange(remoteChange);
        return super.acknowledgeChange(remoteChange, change);
    }

    /**
     * Sets up a bucket instance with the given BucketSchema and provides
     * mock instances of a Bucket's depenencies for testing objects that
     * interface with a bucket.
     */
    public static <T extends Syncable> MockBucket<T> buildBucket(BucketSchema<T> schema)
    throws BucketNameInvalid {
        return buildBucket(schema, new MockChannelProvider());
    }


    /**
     * Sets up a bucket instance with the provided BucketSchema and configures
     * the ChannelProvider to be used with the bucket.
     */
    public static <T extends Syncable> MockBucket<T> buildBucket(BucketSchema<T> schema, ChannelProvider provider)
    throws BucketNameInvalid {
        return buildBucket(MockExecutor.immediate(), schema, provider);
    }

    /**
     * Build a mock bucket with the given Executor, Schema and ChannelProvider
     */
    public static <T extends Syncable> MockBucket<T> buildBucket(Executor executor, BucketSchema<T> schema, ChannelProvider provider)
    throws BucketNameInvalid {
        User user = MockUser.buildUser();
        BucketStore<T> store = new MockBucketStore<T>();
        MockGhostStore ghosts = new MockGhostStore();

        MockBucket<T> bucket = new MockBucket<T>(executor, schema.getRemoteName(), schema, user, store, ghosts);

        Bucket.Channel channel = provider.buildChannel(bucket);
        bucket.setChannel(channel);
        bucket.start();
        return bucket;
    }

} public class MockChannelListener implements Channel.OnMessageListener {

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

    public boolean open = false, autoAcknowledge = false, initReceived = false;
    public String api = null;
    public List<Channel.MessageEvent> messages = Collections.synchronizedList(new ArrayList<Channel.MessageEvent>());
    public List<String> logs = new ArrayList<String>();
    public Channel.MessageEvent lastMessage;
    public Map<String,String> indexData = new HashMap<String,String>();
    public JSONArray indexVersions;
    public Integer replyWithError = null;

    public void sendEmptyIndex(Channel channel){
        channel.receiveMessage("i:{\"index\":[]}");
    }

    public void clearMessages(){
        messages.clear();
        lastMessage = null;
    }

    @Override
    public void onMessage(Channel.MessageEvent event) {
        Log.d(TAG, String.format("Received message: %s", event));
        messages.add(event);

        Message message = parseMessage(event.message);

        if (message.isCommand(Channel.COMMAND_INIT)) {
            // init already received
            if (initReceived) throw new RuntimeException("Channel sent init after already sending init");
            initReceived = true;

            try {
                JSONObject initParams = new JSONObject(message.payload);
                api = initParams.getString(Channel.FIELD_API_VERSION);
            } catch (JSONException e) {
                throw new RuntimeException("Invalid init params", e);
            }

        }

        if (message.isCommand(Channel.COMMAND_ENTITY)) {
            if (indexData.containsKey(message.payload)) {
                Channel channel = (Channel) event.getSource();
                channel.receiveMessage(String.format("e:%s\n%s", message.payload, indexData.get(message.payload)));;
            }
        }

        if (message.isCommand(Channel.COMMAND_CHANGE)) {

            if (replyWithError != null) {
                ChannelUtil.replyWithError(event, replyWithError);
                replyWithError = 409;
            } else if (autoAcknowledge) {
                ChannelUtil.acknowledgeChange(event);
            }

        }

        lastMessage = event;
    }

    @Override
    public void onLog(Channel channel, int level, CharSequence message) {
        Log.d(TAG, String.format("%s (%d): %s", channel.getBucketName(), level, message));
        logs.add(message.toString());
    }

    @Override
    public void onClose(Channel channel) {
        open = false;
    }

    @Override
    public void onOpen(Channel channel) {
        open = true;
    }

    static public final String COLON = ":";
    public static Message parseMessage(String message) {
        int colon = message.indexOf(COLON);
        String command = null, payload = null;
        if (colon == -1) {
            command = message;
            payload = "";
        } else {
            command = message.substring(0, colon);
            payload = message.substring(colon+1);
        }

        return new Message(command, payload);
    }

    public static class Message {

        final public String command;
        final public String payload;

        public Message(String command, String payload) {
            this.command = command;
            this.payload = payload;
        }

        public boolean isCommand(String command) {
            return this.command.equals(command);
        }
    }

} public class MockChannelSerializer implements Serializer {

    public int ackCount = 0;

    public SerializedQueue queue = new SerializedQueue();

    /**
     * Return what we want the queue to start with
     */
    @Override
    public SerializedQueue restore(Bucket bucket){
        return queue;
    }

    /**
     * Create a new blank queue
     */
    @Override
    public void reset(Bucket bucket){
        queue = new SerializedQueue();
    }

    public void onQueueChange(Change change){
        queue.queued.add(change);
    }

    public void onDequeueChange(Change change){
        queue.queued.remove(change);
    }

    public void onSendChange(Change change){
        queue.queued.remove(change);
        queue.pending.put(change.getKey(), change);
    }

    public void onAcknowledgeChange(Change change){
        ackCount ++;
        queue.pending.remove(change.getKey());
    }

} 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 class ChannelUtil {

    public static JSONObject parseChangeData(Channel.MessageEvent event)
    throws JSONException {
        if (event.message.substring(0,2).equals("c:")) {
            JSONObject change = new JSONObject(event.message.substring(2));
            return change;
        }

        throw new RuntimeException("Not a change message: " + event.message);
    }

    /**
     * Send a remote change message to the channel that ransforms Syncable origin
     * to contain target representation.
     * 
     * Generates a random `cv` and increments the `sv` by one.
     */
    public static void sendRemoteChange(Channel channel, Syncable origin, JSONObject target) {
        try {
            String id = origin.getSimperiumKey();
            int version = origin.getVersion();
            JSONObject diff = JSONDiff.diff(origin.getUnmodifiedValue(), target);
            JSONObject modify = RemoteChangesUtil.modifyOperation(id, version, diff.getJSONObject("v"));

            channel.receiveMessage("c:[" + modify.toString() + "]");
        } catch (JSONException e) {
            throw new RuntimeException("Unable to send remote change", e);
        }
    }

    public static void replyWithError(Channel.MessageEvent event, Integer error) {

        try {
            JSONObject change = parseChangeData(event);
            JSONObject reply = new JSONObject();
            reply.put("ccids", new JSONArray("[\"" + change.getString("ccid") + "\"]"));
            reply.put("id", change.getString("id"));
            reply.put("clientid", event.channel.getSessionId());
            reply.put("error", error);

            event.channel.receiveMessage("c:[" + reply.toString() + "]");

        } catch (JSONException e) {
            throw new RuntimeException(String.format("Couldn't reply with error %s", event.message), e);
        }

    }

    public static void acknowledgeChange(Channel.MessageEvent event) {

        try {
            JSONArray responseJSON = RemoteChangesUtil.acknowledgeChange(event);
            event.channel.receiveMessage(String.format("c:%s", responseJSON));
        } catch (JSONException e) {
            throw new RuntimeException(String.format("Couldn't auto-acknowledge %s", event.message), e);
        }

    }

    public static void sendModifyOperation(Channel channel, String objectId, int sourceVersion, JSONObject diff) {

        try {
            JSONObject modify = RemoteChangesUtil.modifyOperation(objectId, sourceVersion, diff);
            channel.receiveMessage(String.format("c:[%s]", modify));
        } catch (JSONException e) {
            throw new RuntimeException("Could not build change", e);
        }

    }

    public static void sendObject(Channel channel, String id, int version, JSONObject data) {
        try {
            JSONObject payload = new JSONObject();
            payload.put("data", data);
            channel.receiveMessage(String.format(Locale.US, "e:%s.%d\n%s", id, version, payload));
        } catch (JSONException e) {
            throw new RuntimeException("Could not send object data", e);
        }
    }

} public class RemoteChangesUtil {

    public static JSONObject operation(String id, String client, String cv, String ccid)
    throws JSONException {

        JSONObject change = new JSONObject();
        change.put(RemoteChange.CHANGE_IDS_KEY, new JSONArray("[\"" + ccid + "\"]"));
        change.put(RemoteChange.CLIENT_KEY, client);
        change.put(RemoteChange.CHANGE_VERSION_KEY, cv);
        change.put(RemoteChange.ID_KEY, id);

        return change;
    }

    public static JSONObject operation(String id, String client, String changeId)
    throws JSONException {
        return operation(id, client, Uuid.uuid(), changeId);
    }

    public static JSONObject operation(String id, String client)
    throws JSONException {
        return operation(id, client, Uuid.uuid());
    }

    public static JSONObject operation(String id)
    throws JSONException {
        return operation(id, "mock-client");
    }

    /**
     * Generate a delete operation for the given object id with a random ccid
     * and cv.
     */
    public static JSONObject deleteOperation(String id)
    throws JSONException {
        
        JSONObject change = operation(id);
        change.put(RemoteChange.OPERATION_KEY, RemoteChange.OPERATION_REMOVE);

        return change;

    }

    public static JSONObject addOperation(String id)
    throws JSONException {
        JSONObject change = operation(id);
        change.put(RemoteChange.OPERATION_KEY, RemoteChange.OPERATION_MODIFY);
        return change;
    }

    public static JSONObject modifyOperation(String id, int sourceVersion, JSONObject diff)
    throws JSONException {
        JSONObject change = addOperation(id);
        change.put(RemoteChange.SOURCE_VERSION_KEY, sourceVersion);
        change.put(RemoteChange.END_VERSION_KEY, sourceVersion + 1);
        change.put(JSONDiff.DIFF_VALUE_KEY, diff);

        return change;
    }

    public static JSONArray acknowledgeChange(Channel.MessageEvent channelEvent)
    throws JSONException {
        JSONObject change = new JSONObject(channelEvent.message.substring(2));
        return acknowledgeChange(change, channelEvent.channel);
    }

    public static JSONArray acknowledgeChange(JSONObject change, Channel channel)
    throws JSONException {

        String id = change.getString(Change.ID_KEY);
        String client = channel.getSessionId();
        String ccid = change.getString(Change.CHANGE_ID_KEY);

        JSONObject ack = operation(id, client, ccid);
        ack.put(RemoteChange.OPERATION_KEY, change.getString(Change.OPERATION_KEY));
        ack.put(JSONDiff.DIFF_VALUE_KEY, change.getJSONObject(JSONDiff.DIFF_VALUE_KEY));

        int sv = -1;

        if (change.has(RemoteChange.SOURCE_VERSION_KEY)) {
            sv = change.getInt(RemoteChange.SOURCE_VERSION_KEY);
            ack.put(RemoteChange.SOURCE_VERSION_KEY, change.getInt(RemoteChange.SOURCE_VERSION_KEY));
        }

        ack.put(RemoteChange.END_VERSION_KEY, sv + 1);

        JSONArray response = new JSONArray();
        response.put(ack);

        return response;
    }

    public static RemoteChange buildRemoteChange(Syncable origin, JSONObject target)
    throws JSONException {

        String id = origin.getSimperiumKey();
        int version = origin.getVersion();
        JSONObject diff = JSONDiff.diff(origin.getUnmodifiedValue(), target);
        JSONObject modify = modifyOperation(id, version, diff.getJSONObject("v"));

        return RemoteChange.buildFromMap(modify);
    }

} public interface Flag {
    abstract boolean isComplete();
} public static void waitUntil(Flag flag, String message, long timeout)
throws InterruptedException {
    long start = System.currentTimeMillis();
    while(!flag.isComplete()){
        Thread.sleep(100);
        if (System.currentTimeMillis() - start > timeout) {
            throw(new InterruptedException(message));
        }
    }
}
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
import com.simperium.client.Bucket;
import com.simperium.client.Change;
import com.simperium.client.Channel;
import com.simperium.client.ChannelProvider;
import com.simperium.client.RemoteChange;
import com.simperium.client.User;
import com.simperium.models.Note;
import com.simperium.test.MockBucket;
import com.simperium.test.MockChannelListener;
import com.simperium.test.MockChannelSerializer;
import com.simperium.test.MockExecutor;
import com.simperium.util.ChannelUtil;
import com.simperium.util.RemoteChangesUtil;
import org.json.JSONArray;
import org.json.JSONObject;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import static android.test.MoreAsserts.assertMatchesRegex;
import static com.simperium.TestHelpers.Flag;
import static com.simperium.TestHelpers.waitUntil;
package com.simperium;





public class ChannelTest extends BaseSimperiumTest {


    public static String SESSION_ID = "SESSION-ID";
    public static String APP_ID = "APP_ID";

    private MockBucket<Note> mBucket;
    private Channel mChannel;