package meerkat.bulletinboard.workers.multiserver;

import com.google.common.util.concurrent.FutureCallback;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import meerkat.bulletinboard.MultiServerWorker;
import meerkat.bulletinboard.SingleServerBulletinBoardClient;
import meerkat.bulletinboard.BatchDataContainer;

import java.util.Iterator;
import java.util.List;

/**
 * Created by Arbel Deutsch Peled on 27-Dec-15.
 */
public class MultiServerPostBatchDataWorker extends MultiServerWorker<BatchDataContainer, Boolean> {

    public MultiServerPostBatchDataWorker(List<SingleServerBulletinBoardClient> clients,
                                          int minServers, BatchDataContainer payload, int maxRetry,
                                          FutureCallback<Boolean> futureCallback) {

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

    }

    @Override
    public void run() {

        Iterator<BatchIdentifier> identifierIterator = payload.batchId.getIdentifiers().iterator();

        // Iterate through client

        for (SingleServerBulletinBoardClient client : clients) {

            if (identifierIterator.hasNext()) {

                // Fetch the batch identifier supplied by the specific client (may be null if batch open failed on client

                BatchIdentifier identifier = identifierIterator.next();

                if (identifier != null) {

                    // Post the data with the matching identifier to the client
                    client.postBatchData(identifier, payload.batchChunkList, payload.startPosition, this);

                } else {

                    // Count servers with no batch identifier as failed
                    maxFailedServers.decrementAndGet();

                }

            }

        }

    }

    @Override
    public void onSuccess(Boolean result) {
        if (minServers.decrementAndGet() <= 0){
            succeed(result);
        }
    }

    @Override
    public void onFailure(Throwable t) {
        if (maxFailedServers.decrementAndGet() <= 0){
            fail(t);
        }
    }

}
