package meerkat.bulletinboard.workers.multiserver;

import com.google.common.util.concurrent.FutureCallback;
import meerkat.bulletinboard.MultiServerBatchIdentifier;
import meerkat.bulletinboard.MultiServerWorker;
import meerkat.bulletinboard.SingleServerBulletinBoardClient;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import meerkat.comm.CommunicationException;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Arbel Deutsch Peled on 27-Dec-15.
 */
public class MultiServerBeginBatchWorker extends MultiServerWorker<Iterable<String>, BatchIdentifier> {

    private BatchIdentifier[] identifiers;
    private AtomicInteger remainingServers;

    public MultiServerBeginBatchWorker(List<SingleServerBulletinBoardClient> clients,
                                       int minServers, Iterable<String> payload, int maxRetry,
                                       FutureCallback<BatchIdentifier> futureCallback) {

        super(clients, minServers, payload, maxRetry, futureCallback);

        identifiers = new BatchIdentifier[clients.size()];

        for (int i = 0 ; i < identifiers.length ; i++) {
            identifiers[i] = null;
        }

        remainingServers = new AtomicInteger(clients.size());

    }

    private class BeginBatchCallback implements FutureCallback<BatchIdentifier> {

        private final int clientNum;

        public BeginBatchCallback(int clientNum) {
            this.clientNum = clientNum;
        }

        private void finishPost() {

            if (remainingServers.decrementAndGet() <= 0){

                if (minServers.decrementAndGet() <= 0) {
                    MultiServerBeginBatchWorker.this.onSuccess(new MultiServerBatchIdentifier(identifiers));
                } else {
                    MultiServerBeginBatchWorker.this.onFailure(new CommunicationException("Could not open batch in enough servers"));
                }
            }

        }

        @Override
        public void onSuccess(BatchIdentifier result) {

            identifiers[clientNum] = result;
            finishPost();

        }

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

    @Override
    public void onSuccess(BatchIdentifier result) {
        succeed(result);
    }

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

    @Override
    public void run() {

        int clientNum = 0;

        for (SingleServerBulletinBoardClient client : clients){

            client.beginBatch(payload, new BeginBatchCallback(clientNum));

            clientNum++;

        }

    }
}
