package meerkat.bulletinboard;

import com.google.common.util.concurrent.FutureCallback;

import com.google.protobuf.Timestamp;
import meerkat.bulletinboard.workers.multiserver.*;
import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.protobuf.Crypto.Signature;
import meerkat.protobuf.Voting.*;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by Arbel Deutsch Peled on 05-Dec-15.
 * Thread-based implementation of a Async Bulletin Board Client.
 * Features:
 * 1. Handles tasks concurrently.
 * 2. Retries submitting
 */
public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient implements AsyncBulletinBoardClient {

    // Executor service for handling jobs
    private final static int JOBS_THREAD_NUM = 5;
    private ExecutorService executorService;

    // Per-server clients
    private List<SingleServerBulletinBoardClient> clients;

    private BulletinBoardDigest batchDigest;

    private final static int POST_MESSAGE_RETRY_NUM = 3;
    private final static int READ_MESSAGES_RETRY_NUM = 1;
    private final static int GET_REDUNDANCY_RETRY_NUM = 1;

    private final int SERVER_THREADPOOL_SIZE;
    private final long FAIL_DELAY;
    private final long SUBSCRIPTION_INTERVAL;

    private static final int DEFAULT_SERVER_THREADPOOL_SIZE = 5;
    private static final long DEFAULT_FAIL_DELAY = 5000;
    private static final long DEFAULT_SUBSCRIPTION_INTERVAL = 10000;

    private int minAbsoluteRedundancy;


    public ThreadedBulletinBoardClient(int serverThreadpoolSize, long failDelay, long subscriptionInterval) {
        SERVER_THREADPOOL_SIZE = serverThreadpoolSize;
        FAIL_DELAY = failDelay;
        SUBSCRIPTION_INTERVAL = subscriptionInterval;
    }

    public ThreadedBulletinBoardClient() {
        this(DEFAULT_SERVER_THREADPOOL_SIZE, DEFAULT_FAIL_DELAY, DEFAULT_SUBSCRIPTION_INTERVAL);
    }

    /**
     * Stores database locations and initializes the web Client
     * Stores the required minimum redundancy.
     * Starts the Thread Pool.
     * @param clientParams contains the required information
     */
    @Override
    public void init(BulletinBoardClientParams clientParams) {

        super.init(clientParams);

        batchDigest = new GenericBulletinBoardDigest(digest);

        minAbsoluteRedundancy = (int) (clientParams.getMinRedundancy() * (float) clientParams.getBulletinBoardAddressCount());

        executorService = Executors.newFixedThreadPool(JOBS_THREAD_NUM);

        clients = new ArrayList<>(clientParams.getBulletinBoardAddressCount());
        for (String address : clientParams.getBulletinBoardAddressList()){

            SingleServerBulletinBoardClient client =
                    new SingleServerBulletinBoardClient(SERVER_THREADPOOL_SIZE, FAIL_DELAY, SUBSCRIPTION_INTERVAL);

            client.init(BulletinBoardClientParams.newBuilder()
                            .addBulletinBoardAddress(address)
                            .build());

            clients.add(client);

        }

    }

    /**
     * Post message to all DBs
     * Retry failed DBs
     * @param msg is the message,
     * @return the message ID for later retrieval
     */
    @Override
    public MessageID postMessage(BulletinBoardMessage msg, FutureCallback<Boolean> callback){

        // Create job
        MultiServerPostMessageWorker worker =
                new MultiServerPostMessageWorker(clients, minAbsoluteRedundancy, msg, POST_MESSAGE_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

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

    }

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

        // Create job
        MultiServerPostBatchWorker worker =
                new MultiServerPostBatchWorker(clients, minAbsoluteRedundancy, msg, chunkSize, POST_MESSAGE_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

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

    }

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

        // Create job
        MultiServerBeginBatchWorker worker =
                new MultiServerBeginBatchWorker(clients, minAbsoluteRedundancy, tags, POST_MESSAGE_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

    }

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

        // Cast identifier to usable form

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

        MultiServerBatchIdentifier identifier = (MultiServerBatchIdentifier) batchIdentifier;

        BatchDataContainer batchDataContainer = new BatchDataContainer(identifier, batchChunkList, startPosition);

        // Create job
        MultiServerPostBatchDataWorker worker =
                new MultiServerPostBatchDataWorker(clients, minAbsoluteRedundancy, batchDataContainer, POST_MESSAGE_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

    }

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

        postBatchData(batchIdentifier, batchChunkList, 0, callback);

    }

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

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

        MultiServerBatchIdentifier identifier = (MultiServerBatchIdentifier) payload;

        // Create job
        MultiServerCloseBatchWorker worker =
                new MultiServerCloseBatchWorker(clients, minAbsoluteRedundancy, identifier, timestamp, signatures, POST_MESSAGE_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

    }

    /**
     * Access each database and search for a given message ID
     * Return the number of databases in which the message was found
     * Only try once per DB
     * Ignore communication exceptions in specific databases
     */
    @Override
    public void getRedundancy(MessageID id, FutureCallback<Float> callback) {

        // Create job
        MultiServerGetRedundancyWorker worker =
                new MultiServerGetRedundancyWorker(clients, minAbsoluteRedundancy, id, GET_REDUNDANCY_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

    }

    /**
     * Go through the DBs and try to retrieve messages according to the specified filter
     * If at the operation is successful for some DB: return the results and stop iterating
     * If no operation is successful: return null (NOT blank list)
     */
    @Override
    public void readMessages(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {

        // Create job
        MultiServerReadMessagesWorker worker =
                new MultiServerReadMessagesWorker(clients, minAbsoluteRedundancy, filterList, READ_MESSAGES_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

    }

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

        //Create job
        MultiServerReadMessageWorker worker =
                new MultiServerReadMessageWorker(clients, minAbsoluteRedundancy, msgID, READ_MESSAGES_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

    }

    @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
        MultiServerReadBatchDataWorker worker =
                new MultiServerReadBatchDataWorker(clients, minAbsoluteRedundancy, stub, READ_MESSAGES_RETRY_NUM, callback);

        // Submit job
        executorService.submit(worker);

    }

    /**
     * This method is not supported by this class!
     * This is because it has no meaning when considering more than one server without knowing which server will be contacted
     */
    @Override
    public void querySync(SyncQuery syncQuery, FutureCallback<SyncQueryResponse> callback) {
        callback.onFailure(new IllegalAccessError("querySync is not supported by this class"));
    }

    @Override
    public void close() {
        super.close();

        try {

            for (SingleServerBulletinBoardClient client : clients){
                client.close();
            }

            executorService.shutdown();
            while (! executorService.isShutdown()) {
                executorService.awaitTermination(10, TimeUnit.SECONDS);
            }
        } catch (InterruptedException e) {
            System.err.println(e.getCause() + " " + e.getMessage());
        }
    }

}
