package meerkat.bulletinboard;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.Int64Value;
import com.google.protobuf.Timestamp;
import meerkat.bulletinboard.workers.singleserver.*;
import meerkat.comm.CommunicationException;
import meerkat.crypto.concrete.SHA256Digest;
import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.protobuf.Crypto;
import meerkat.protobuf.Voting.BulletinBoardClientParams;
import meerkat.util.BulletinBoardUtils;

import javax.ws.rs.client.Client;
import java.lang.Iterable;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Arbel Deutsch Peled on 28-Dec-15.
 *
 * This class implements the asynchronous Bulletin Board Client interface
 * It only handles a single Bulletin Board Server
 * If the list of servers contains more than one server: the server actually used is the first one
 * The class further implements a delayed access to the server after a communication error occurs
 */
public class SingleServerBulletinBoardClient implements SubscriptionBulletinBoardClient {

    protected Client client;

    protected BulletinBoardDigest digest;

    private String dbAddress;

    private final int MAX_RETRIES = 11;

    private final ListeningScheduledExecutorService executorService;

    private long lastServerErrorTime;

    private final long FAIL_DELAY_IN_MILLISECONDS;

    private final long SUBSCRIPTION_INTERVAL_IN_MILLISECONDS;

    /**
     * Notify the client that a job has failed
     * This makes new scheduled jobs be scheduled for a later time (after the given delay)
     */
    protected void fail() {

        // Update last fail time
        lastServerErrorTime = System.currentTimeMillis();

    }

    private class SynchronousRetry<OUT> {

        private final SingleServerWorker<?,OUT> worker;

        private String thrown;

        public SynchronousRetry(SingleServerWorker<?,OUT> worker) {
            this.worker = worker;
            this.thrown = "Could not contact server. Errors follow:\n";
        }

        OUT run() throws CommunicationException {

            do {

                try {
                    return worker.call();
                } catch (Exception e) {
                    thrown += e.getCause() + " " + e.getMessage() + "\n";
                }

                try {
                    Thread.sleep(FAIL_DELAY_IN_MILLISECONDS);
                } catch (InterruptedException e) {
                    //TODO: log
                }

                worker.decMaxRetry();

            } while (worker.isRetry());

            throw new CommunicationException(thrown);

        }

    }

    /**
     * This method adds a worker to the scheduled queue of the threadpool
     * If the server is in an accessible state: the job is submitted for immediate handling
     * If the server is not accessible: the job is scheduled for a later time
     * @param worker is the worker that should be scheduled for work
     * @param callback is the class containing callbacks for handling job completion/failure
     */
    protected void scheduleWorker(SingleServerWorker worker, FutureCallback callback){

        long timeSinceLastServerError = System.currentTimeMillis() - lastServerErrorTime;

        if (timeSinceLastServerError >= FAIL_DELAY_IN_MILLISECONDS) {

            // Schedule for immediate processing
            Futures.addCallback(executorService.submit(worker), callback);

        } else {

            // Schedule for processing immediately following delay expiry
            Futures.addCallback(executorService.schedule(
                    worker,
                    FAIL_DELAY_IN_MILLISECONDS - timeSinceLastServerError,
                    TimeUnit.MILLISECONDS),
                    callback);

        }

    }

    /**
     * Inner class for handling simple operation results and retrying if needed
     */
    class RetryCallback<T> implements FutureCallback<T> {

        private final SingleServerWorker worker;
        private final FutureCallback<T> futureCallback;

        public RetryCallback(SingleServerWorker worker, FutureCallback<T> futureCallback) {
            this.worker = worker;
            this.futureCallback = futureCallback;
        }

        @Override
        public void onSuccess(T result) {
            if (futureCallback != null)
                futureCallback.onSuccess(result);
        }

        @Override
        public void onFailure(Throwable t) {

            // Notify client about failure
            fail();

            // Check if another attempt should be made

            worker.decMaxRetry();

            if (worker.isRetry()) {
                // Perform another attempt
                scheduleWorker(worker, this);
            } else {
                // No more retries: notify caller about failure
                if (futureCallback != null)
                    futureCallback.onFailure(t);
            }

        }

    }

    /**
     * This callback ties together all the per-batch-data callbacks into a single callback
     * It reports success back to the user only if all of the batch-data were successfully posted
     * If any batch-data fails to post: this callback reports failure
     */
    class PostBatchChunkListCallback implements FutureCallback<Boolean> {

        private final FutureCallback<Boolean> callback;

        private AtomicInteger batchDataRemaining;
        private AtomicBoolean aggregatedResult;

        public PostBatchChunkListCallback(int batchDataLength, FutureCallback<Boolean> callback) {

            this.callback = callback;
            this.batchDataRemaining = new AtomicInteger(batchDataLength);
            this.aggregatedResult = new AtomicBoolean(false);

        }

        @Override
        public void onSuccess(Boolean result) {

            if (result){
                this.aggregatedResult.set(true);
            }

            if (batchDataRemaining.decrementAndGet() == 0){
                if (callback != null)
                    callback.onSuccess(this.aggregatedResult.get());
            }
        }

        @Override
        public void onFailure(Throwable t) {

            // Notify caller about failure
            if (callback != null)
                callback.onFailure(t);

        }
    }

    private class ReadBatchCallback implements FutureCallback<List<BatchChunk>> {

        private final BulletinBoardMessage stub;
        private final FutureCallback<BulletinBoardMessage> callback;

        public ReadBatchCallback(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) {
            this.stub = stub;
            this.callback = callback;
        }

        @Override
        public void onSuccess(List<BatchChunk> result) {
            callback.onSuccess(BulletinBoardUtils.gatherBatch(stub, result));
        }

        @Override
        public void onFailure(Throwable t) {
            callback.onFailure(t);
        }

    }

    /**
     * This callback receives a message which may be a stub
     * If the message is not a stub: it returns it as is to a callback function
     * If it is a stub: it schedules a read of the batch data which will return a complete message to the callback function
     */
    class CompleteMessageReadCallback implements FutureCallback<List<BulletinBoardMessage>>{

        private final FutureCallback<BulletinBoardMessage> callback;

        public CompleteMessageReadCallback(FutureCallback<BulletinBoardMessage> callback) {

            this.callback = callback;

        }

        @Override
        public void onSuccess(List<BulletinBoardMessage> result) {
            if (result.size() <= 0) {
                onFailure(new CommunicationException("Could not find required message on the server."));
            } else {

                BulletinBoardMessage msg = result.get(0);

                if (msg.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {
                    callback.onSuccess(msg);
                } else {

                    // Create job with MAX retries for retrieval of the Batch Data List

                    BatchQuery batchQuery = BatchQuery.newBuilder()
                            .setMsgID(MessageID.newBuilder()
                                    .setID(msg.getMsg().getMsgId())
                                    .build())
                            .build();

                    SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(dbAddress, batchQuery, MAX_RETRIES);

                    scheduleWorker(batchWorker, new ReadBatchCallback(msg, callback));

                }
            }
        }

        @Override
        public void onFailure(Throwable t) {
            callback.onFailure(t);
        }

    }


    /**
     * Inner class for handling returned values of subscription operations
     * This class's methods also ensure continued operation of the subscription
     */
    class SubscriptionCallback implements FutureCallback<List<BulletinBoardMessage>> {

        private SingleServerReadMessagesWorker worker;
        private final FutureCallback<List<BulletinBoardMessage>> callback;

        private MessageFilterList.Builder filterBuilder;

        public SubscriptionCallback(SingleServerReadMessagesWorker worker, FutureCallback<List<BulletinBoardMessage>> callback) {
            this.worker = worker;
            this.callback = callback;
            filterBuilder = worker.getPayload().toBuilder();

        }

        @Override
        public void onSuccess(List<BulletinBoardMessage> result) {

            // Report new messages to user
            if (callback != null)
                callback.onSuccess(result);

            // Update filter if needed

            if (result.size() > 0) {

                // Remove last filter from list (MIN_ENTRY one)
                filterBuilder.removeFilter(filterBuilder.getFilterCount() - 1);

                // Add updated MIN_ENTRY filter (entry number is successor of last received entry's number)
                filterBuilder.addFilter(MessageFilter.newBuilder()
                        .setType(FilterType.MIN_ENTRY)
                        .setEntry(result.get(result.size() - 1).getEntryNum() + 1)
                        .build());

            }

            // Create new worker with updated task
            worker = new SingleServerReadMessagesWorker(worker.serverAddress, filterBuilder.build(), MAX_RETRIES);

            RetryCallback<List<BulletinBoardMessage>> retryCallback = new RetryCallback<>(worker, this);

            // Schedule the worker to run after the given interval has elapsed
            Futures.addCallback(executorService.schedule(worker, SUBSCRIPTION_INTERVAL_IN_MILLISECONDS, TimeUnit.MILLISECONDS), retryCallback);

        }

        @Override
        public void onFailure(Throwable t) {

            // Notify client about failure
            fail();

            // Notify caller about failure and terminate subscription
            if (callback != null)
                callback.onFailure(t);
        }
    }

    public SingleServerBulletinBoardClient(ListeningScheduledExecutorService executorService,
                                           long failDelayInMilliseconds,
                                           long subscriptionIntervalInMilliseconds) {

        this.executorService = executorService;

        this.FAIL_DELAY_IN_MILLISECONDS = failDelayInMilliseconds;
        this.SUBSCRIPTION_INTERVAL_IN_MILLISECONDS = subscriptionIntervalInMilliseconds;

        // Set server error time to a time sufficiently in the past to make new jobs go through
        lastServerErrorTime = System.currentTimeMillis() - failDelayInMilliseconds;

    }

    public SingleServerBulletinBoardClient(int threadPoolSize, long failDelayInMilliseconds, long subscriptionIntervalInMilliseconds) {

        this(MoreExecutors.listeningDecorator(Executors.newScheduledThreadPool(threadPoolSize)),
                failDelayInMilliseconds,
                subscriptionIntervalInMilliseconds);

    }

    /**
     * Stores database location, initializes the web Client and
     * @param clientParams contains the data needed to access the DBs
     */
    @Override
    public void init(BulletinBoardClientParams clientParams) {

        this.digest = new GenericBulletinBoardDigest(new SHA256Digest());

        // Remove all but first DB address
        this.dbAddress = clientParams.getBulletinBoardAddress(0);

    }

    // Synchronous methods

    @Override
    public MessageID postMessage(BulletinBoardMessage msg) throws CommunicationException {

        SingleServerPostMessageWorker worker = new SingleServerPostMessageWorker(dbAddress, msg, MAX_RETRIES);

        SynchronousRetry<Boolean> retry = new SynchronousRetry<>(worker);

        retry.run();

        digest.reset();
        digest.update(msg);

        return digest.digestAsMessageID();

    }

    @Override
    public float getRedundancy(MessageID id) throws CommunicationException {

        SingleServerGetRedundancyWorker worker = new SingleServerGetRedundancyWorker(dbAddress, id, MAX_RETRIES);

        SynchronousRetry<Float> retry = new SynchronousRetry<>(worker);

        return retry.run();

    }

    @Override
    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException {

        SingleServerReadMessagesWorker worker = new SingleServerReadMessagesWorker(dbAddress, filterList, MAX_RETRIES);

        SynchronousRetry<List<BulletinBoardMessage>> retry = new SynchronousRetry<>(worker);

        return retry.run();

    }

    @Override
    public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize) throws CommunicationException {

        // Begin the batch and obtain identifier

        BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder()
                .addAllTag(msg.getMsg().getTagList())
                .build();

        SingleServerBeginBatchWorker beginBatchWorker = new SingleServerBeginBatchWorker(dbAddress, beginBatchMessage, MAX_RETRIES);

        SynchronousRetry<Int64Value> beginRetry = new SynchronousRetry<>(beginBatchWorker);

        Int64Value identifier = beginRetry.run();

        // Post data chunks

        List<BatchChunk> batchChunkList = BulletinBoardUtils.breakToBatch(msg, chunkSize);

        BatchMessage.Builder builder = BatchMessage.newBuilder().setBatchId(identifier.getValue());

        int position = 0;

        for (BatchChunk data : batchChunkList) {

            builder.setSerialNum(position).setData(data);

            SingleServerPostBatchWorker dataWorker = new SingleServerPostBatchWorker(dbAddress, builder.build(), MAX_RETRIES);

            SynchronousRetry<Boolean> dataRetry = new SynchronousRetry<>(dataWorker);

            dataRetry.run();

            // Increment position in batch
            position++;

        }

        // Close batch

        CloseBatchMessage closeBatchMessage = CloseBatchMessage.newBuilder()
                .setBatchId(identifier.getValue())
                .addAllSig(msg.getSigList())
                .setTimestamp(msg.getMsg().getTimestamp())
                .setBatchLength(position)
                .build();

        SingleServerCloseBatchWorker closeBatchWorker = new SingleServerCloseBatchWorker(dbAddress, closeBatchMessage, MAX_RETRIES);

        SynchronousRetry<Boolean> retry = new SynchronousRetry<>(closeBatchWorker);

        retry.run();

        // Calculate ID and return

        digest.reset();
        digest.update(msg);

        return digest.digestAsMessageID();

    }

    @Override
    public BulletinBoardMessage readMessage(MessageID msgID) throws CommunicationException {

        // Retrieve message (which may be a stub)

        MessageFilterList filterList = MessageFilterList.newBuilder()
                .addFilter(MessageFilter.newBuilder()
                        .setType(FilterType.MSG_ID)
                        .setId(msgID.getID())
                        .build())
                .build();

        SingleServerReadMessagesWorker stubWorker = new SingleServerReadMessagesWorker(dbAddress, filterList, MAX_RETRIES);

        SynchronousRetry<List<BulletinBoardMessage>> retry = new SynchronousRetry<>(stubWorker);

        List<BulletinBoardMessage> messages = retry.run();

        if (messages.size() <= 0) {
            throw new CommunicationException("Could not find message in database.");
        }

        BulletinBoardMessage msg = messages.get(0);

        if (msg.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {

            // We retrieved a complete message. Return it.
            return msg;

        } else {

            // We retrieved a stub. Retrieve data.
            return readBatchData(msg);

        }

    }

    @Override
    public BulletinBoardMessage readBatchData(BulletinBoardMessage stub) throws CommunicationException, IllegalArgumentException {

        BatchQuery batchQuery = BatchQuery.newBuilder()
                .setMsgID(MessageID.newBuilder()
                        .setID(stub.getMsg().getMsgId())
                        .build())
                .setStartPosition(0)
                .build();

        SingleServerReadBatchWorker readBatchWorker = new SingleServerReadBatchWorker(dbAddress, batchQuery, MAX_RETRIES);

        SynchronousRetry<List<BatchChunk>> batchRetry = new SynchronousRetry<>(readBatchWorker);

        List<BatchChunk> batchChunkList = batchRetry.run();

        return BulletinBoardUtils.gatherBatch(stub, batchChunkList);

    }

    @Override
    public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException {

        SingleServerGenerateSyncQueryWorker worker =
                new SingleServerGenerateSyncQueryWorker(dbAddress, generateSyncQueryParams, MAX_RETRIES);

        SynchronousRetry<SyncQuery> retry = new SynchronousRetry<>(worker);

        return retry.run();

    }

    // Asynchronous methods

    @Override
    public MessageID postMessage(BulletinBoardMessage msg, FutureCallback<Boolean> callback) {

        // Create worker with redundancy 1 and MAX_RETRIES retries
        SingleServerPostMessageWorker worker = new SingleServerPostMessageWorker(dbAddress, msg, MAX_RETRIES);

        // Submit worker and create callback
        scheduleWorker(worker, new RetryCallback<>(worker, callback));

        // Calculate the correct message ID and return it
        digest.reset();
        digest.update(msg.getMsg());
        return digest.digestAsMessageID();

    }

    private class PostBatchDataCallback implements FutureCallback<Boolean> {

        private final BulletinBoardMessage msg;
        private final BatchIdentifier identifier;
        private final FutureCallback<Boolean> callback;

        public PostBatchDataCallback(BulletinBoardMessage msg, BatchIdentifier identifier, FutureCallback<Boolean> callback) {
            this.msg = msg;
            this.identifier = identifier;
            this.callback = callback;
        }

        @Override
        public void onSuccess(Boolean result) {
            closeBatch(
                    identifier,
                    msg.getMsg().getTimestamp(),
                    msg.getSigList(),
                    callback
            );
        }

        @Override
        public void onFailure(Throwable t) {
            if (callback != null)
                callback.onFailure(t);
        }

    }

    private class ContinueBatchCallback implements FutureCallback<BatchIdentifier> {

        private final BulletinBoardMessage msg;
        private final int chunkSize;
        private final FutureCallback<Boolean> callback;

        public ContinueBatchCallback(BulletinBoardMessage msg, int chunkSize, FutureCallback<Boolean> callback) {
            this.msg = msg;
            this.chunkSize = chunkSize;
            this.callback = callback;
        }

        @Override
        public void onSuccess(BatchIdentifier identifier) {

            List<BatchChunk> batchChunkList = BulletinBoardUtils.breakToBatch(msg, chunkSize);

            postBatchData(
                    identifier,
                    batchChunkList,
                    0,
                    new PostBatchDataCallback(msg, identifier, callback));
        }

        @Override
        public void onFailure(Throwable t) {
            if (callback != null)
                callback.onFailure(t);
        }
    }

    @Override
    public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback<Boolean> callback) {

        beginBatch(
                msg.getMsg().getTagList(),
                new ContinueBatchCallback(msg, chunkSize, callback)
        );

        digest.update(msg);

        return digest.digestAsMessageID();

    }

    private class BeginBatchCallback implements FutureCallback<Int64Value> {

        private final FutureCallback<BatchIdentifier> callback;

        public BeginBatchCallback(FutureCallback<BatchIdentifier> callback) {
            this.callback = callback;
        }

        @Override
        public void onSuccess(Int64Value result) {
            callback.onSuccess(new SingleServerBatchIdentifier(result));
        }

        @Override
        public void onFailure(Throwable t) {
            callback.onFailure(t);
        }
    }

    @Override
    public void beginBatch(Iterable<String> tags, FutureCallback<BatchIdentifier> callback) {

        BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder()
                .addAllTag(tags)
                .build();

        // Create worker with redundancy 1 and MAX_RETRIES retries
        SingleServerBeginBatchWorker worker =
                new SingleServerBeginBatchWorker(dbAddress, beginBatchMessage, MAX_RETRIES);

        // Submit worker and create callback
        scheduleWorker(worker, new RetryCallback<>(worker, new BeginBatchCallback(callback)));

    }


    @Override
    public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList,
                          int startPosition, FutureCallback<Boolean> callback) throws IllegalArgumentException{

        // Cast identifier to usable form

        if (!(batchIdentifier instanceof SingleServerBatchIdentifier)){
            throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
        }

        SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchIdentifier;

        // Update batch size

        identifier.setLength(startPosition + batchChunkList.size());

        // Create a unified callback to aggregate successful posts

        PostBatchChunkListCallback listCallback = new PostBatchChunkListCallback(batchChunkList.size(), callback);

        // Iterate through data list

        BatchMessage.Builder builder = BatchMessage.newBuilder()
                .setBatchId(identifier.getBatchId().getValue());

        for (BatchChunk data : batchChunkList) {
            builder.setSerialNum(startPosition).setData(data);

            // Create worker with redundancy 1 and MAX_RETRIES retries
            SingleServerPostBatchWorker worker =
                    new SingleServerPostBatchWorker(dbAddress, builder.build(), MAX_RETRIES);

            // Create worker with redundancy 1 and MAX_RETRIES retries
            scheduleWorker(worker, new RetryCallback<>(worker, listCallback));

            // Increment position in batch
            startPosition++;
        }

    }

    @Override
    public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback)
        throws IllegalArgumentException {

        postBatchData(batchIdentifier, batchChunkList, 0, callback);

    }

    @Override
    public void closeBatch(BatchIdentifier batchIdentifier, Timestamp timestamp, Iterable<Crypto.Signature> signatures, FutureCallback<Boolean> callback)
            throws IllegalArgumentException {

        if (!(batchIdentifier instanceof SingleServerBatchIdentifier)){
            throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
        }

        SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchIdentifier;

        CloseBatchMessage closeBatchMessage = CloseBatchMessage.newBuilder()
                .setBatchId(identifier.getBatchId().getValue())
                .setBatchLength(identifier.getLength())
                .setTimestamp(timestamp)
                .addAllSig(signatures)
                .build();

        // Create worker with redundancy 1 and MAX_RETRIES retries
        SingleServerCloseBatchWorker worker =
                new SingleServerCloseBatchWorker(dbAddress, closeBatchMessage, MAX_RETRIES);

        // Submit worker and create callback
        scheduleWorker(worker, new RetryCallback<>(worker, callback));

    }

    @Override
    public void getRedundancy(MessageID id, FutureCallback<Float> callback) {

        // Create worker with no retries
        SingleServerGetRedundancyWorker worker = new SingleServerGetRedundancyWorker(dbAddress, id, 1);

        // Submit job and create callback
        scheduleWorker(worker, new RetryCallback<>(worker, callback));

    }

    @Override
    public void readMessages(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {

        // Create job with no retries
        SingleServerReadMessagesWorker worker = new SingleServerReadMessagesWorker(dbAddress, filterList, 1);

        // Submit job and create callback
        scheduleWorker(worker, new RetryCallback<>(worker, callback));

    }

    @Override
    public void readMessage(MessageID msgID, FutureCallback<BulletinBoardMessage> callback) {

        // Create job with MAX retries for retrieval of the Bulletin Board Message (which may be a stub)

        MessageFilterList filterList = MessageFilterList.newBuilder()
                .addFilter(MessageFilter.newBuilder()
                        .setType(FilterType.MSG_ID)
                        .setId(msgID.getID())
                        .build())
                .build();

        BatchQuery batchQuery = BatchQuery.newBuilder()
                .setMsgID(msgID)
                .setStartPosition(0)
                .build();

        SingleServerReadMessagesWorker messageWorker = new SingleServerReadMessagesWorker(dbAddress, filterList, MAX_RETRIES);

        // Submit jobs with wrapped callbacks
        scheduleWorker(messageWorker, new RetryCallback<>(messageWorker, new CompleteMessageReadCallback(callback)));

    }

    @Override
    public void readBatchData(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) throws IllegalArgumentException{

        if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {
            throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
        }

        // Create job with MAX retries for retrieval of the Batch Data List

        BatchQuery batchQuery = BatchQuery.newBuilder()
                .setMsgID(MessageID.newBuilder()
                        .setID(stub.getMsg().getMsgId())
                        .build())
                .setStartPosition(0)
                .build();

        SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(dbAddress, batchQuery, MAX_RETRIES);

        scheduleWorker(batchWorker, new RetryCallback<>(batchWorker, new ReadBatchCallback(stub, callback)));

    }

    @Override
    public void querySync(SyncQuery syncQuery, FutureCallback<SyncQueryResponse> callback) {

        SingleServerQuerySyncWorker worker = new SingleServerQuerySyncWorker(dbAddress, syncQuery, MAX_RETRIES);

        scheduleWorker(worker, new RetryCallback<>(worker, callback));

    }

    @Override
    public void subscribe(MessageFilterList filterList, long startEntry, FutureCallback<List<BulletinBoardMessage>> callback) {
        // Remove all existing MIN_ENTRY filters and create new one that starts at 0

        MessageFilterList.Builder filterListBuilder = filterList.toBuilder();

        Iterator<MessageFilter> iterator = filterListBuilder.getFilterList().iterator();
        while (iterator.hasNext()) {
            MessageFilter filter = iterator.next();

            if (filter.getType() == FilterType.MIN_ENTRY){
                iterator.remove();
            }
        }
        filterListBuilder.addFilter(MessageFilter.newBuilder()
                .setType(FilterType.MIN_ENTRY)
                .setEntry(startEntry)
                .build());

        // Create job with no retries
        SingleServerReadMessagesWorker worker = new SingleServerReadMessagesWorker(dbAddress, filterListBuilder.build(), MAX_RETRIES);

        // Submit job and create callback that retries on failure and handles repeated subscription
        scheduleWorker(worker, new RetryCallback<>(worker, new SubscriptionCallback(worker, callback)));
    }

    @Override
    public void subscribe(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {
        subscribe(filterList, 0, callback);
    }

    @Override
    public void close() {

        executorService.shutdown();

    }
}
