package meerkat.bulletinboard;

import com.google.common.util.concurrent.FutureCallback;
import com.google.protobuf.Timestamp;
import meerkat.comm.CommunicationException;
import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.util.BulletinBoardUtils;

import static meerkat.protobuf.BulletinBoardApi.FilterType.*;

import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by Arbel Deutsch Peled on 03-Mar-16.
 * A multi-server implementation of the {@link BulletinBoardSubscriber}
 */
public class ThreadedBulletinBoardSubscriber implements BulletinBoardSubscriber {

    protected final Collection<SubscriptionBulletinBoardClient> clients;
    protected final BulletinBoardClient localClient;

    protected Iterator<SubscriptionBulletinBoardClient> clientIterator;
    protected SubscriptionBulletinBoardClient currentClient;

    private long lastServerSwitchTime;

    private AtomicBoolean isSyncInProgress;
    private Semaphore rescheduleSemaphore;

    private AtomicBoolean stopped;

    private static final Float[] BREAKPOINTS = {0.5f, 0.75f, 0.9f, 0.95f, 0.99f, 0.999f};

    public ThreadedBulletinBoardSubscriber(Collection<SubscriptionBulletinBoardClient> clients, BulletinBoardClient localClient) {

        this.clients = clients;
        this.localClient = localClient;

        lastServerSwitchTime = System.currentTimeMillis();

        clientIterator = clients.iterator();
        currentClient = clientIterator.next();

        isSyncInProgress = new AtomicBoolean(false);
        rescheduleSemaphore = new Semaphore(1);

        stopped = new AtomicBoolean(false);

    }

    /**
     * Moves to next client and performs resync with it
     */
    private void nextClient() {

        try {

            rescheduleSemaphore.acquire();

            if (!clientIterator.hasNext()){
                clientIterator = clients.iterator();
            }

            currentClient = clientIterator.next();

            lastServerSwitchTime = System.currentTimeMillis();

            isSyncInProgress.set(false);

            rescheduleSemaphore.release();

        } catch (InterruptedException e) {
            // TODO: log
            // Do not change client
        }

    }

    private abstract class SubscriberCallback<T> implements FutureCallback<T> {

        protected final MessageFilterList filterList;
        protected final FutureCallback<List<BulletinBoardMessage>> callback;
        private final long invocationTime;

        public SubscriberCallback(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {

            this.filterList = filterList;
            this.callback = callback;
            this.invocationTime = System.currentTimeMillis();

        }

        /**
         * Handles resyncing process for the given subscription after a server is switched
         * Specifically: generates a sync query from the local database and uses it to query the current server
         */
        private void reSync() {

            SyncQuery syncQuery = null;
            try {

                syncQuery = localClient.generateSyncQuery(GenerateSyncQueryParams.newBuilder()
                        .setFilterList(filterList)
                        .addAllBreakpointList(Arrays.asList(BREAKPOINTS))
                        .build());

            } catch (CommunicationException e) {

                // Handle failure in standard way
                onFailure(e);

            }

            currentClient.querySync(syncQuery, new SyncQueryCallback(filterList, callback));

        }

        /**
         * Reschedules the subscription
         */
        private void reschedule() {

            try {

                rescheduleSemaphore.acquire();

                reSync();

                rescheduleSemaphore.release();


            } catch (InterruptedException e) {

                //TODO: log

                if (callback != null)
                    callback.onFailure(e); // Hard error: Cannot guarantee subscription safety

            }

        }

        @Override
        public void onFailure(Throwable t) {

            // If server failure is not already known: switch to next client and resync
            if (invocationTime > lastServerSwitchTime){

                // Make sure only what thread switches the client
                if (isSyncInProgress.compareAndSet(false, true)){
                    nextClient();
                }

            }

            reschedule();

        }

    }

    /**
     * Provides handling logic for resync query callback operation
     * Receives a SyncQueryResponse and reads the missing data (starting from the received timestamp) if needed
     */
    protected class SyncQueryCallback extends SubscriberCallback<SyncQueryResponse> {

        public SyncQueryCallback (MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {

            super(filterList, callback);

        }

        @Override
        public void onSuccess(SyncQueryResponse result) {

            final Timestamp DEFAULT_TIME = BulletinBoardUtils.toTimestampProto(946728000); // Year 2000

            // Read required messages according to received Timestamp

            Timestamp syncTimestamp;

            if (result.hasLastTimeOfSync()) {
                syncTimestamp = result.getLastTimeOfSync();                         // Use returned time of sync
            } else {
                syncTimestamp = DEFAULT_TIME;                                       // Get all messages
            }

            MessageFilterList timestampedFilterList = filterList.toBuilder()
                    .removeFilter(filterList.getFilterCount()-1)                    // Remove MIN_ENTRY filter
                    .addFilter(MessageFilter.newBuilder()                           // Add timestamp filter
                            .setType(AFTER_TIME)
                            .setTimestamp(syncTimestamp)
                            .build())
                    .build();

            currentClient.readMessages(timestampedFilterList, new ReSyncCallback(filterList, callback, result.getLastEntryNum()));

        }

    }

    /**
     * Provides handling logic for callback of resyncing process
     * Receives the missing messages, handles them and resubscribes
     */
    protected class ReSyncCallback extends SubscriberCallback<List<BulletinBoardMessage>> {

        private long minEntry;

        public ReSyncCallback (MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback, long minEntry) {

            super(filterList, callback);

            this.minEntry = minEntry;

        }

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

            // Propagate result to caller
            if (callback != null)
                callback.onSuccess(result);

            // Renew subscription

            MessageFilterList newFilterList = filterList.toBuilder()
                    .removeFilter(filterList.getFilterCount()-1)        // Remove current MIN_ENTRY filter
                    .addFilter(MessageFilter.newBuilder()               // Add new MIN_ENTRY filter for current server
                            .setType(MIN_ENTRY)
                            .setEntry(minEntry)
                            .build())
                    .build();

            currentClient.subscribe(newFilterList, callback);

        }

    }

    /**
     * Provides the handling logic for results and failures of main subscription (while there are no errors)
     */
    protected class SubscriptionCallback extends SubscriberCallback<List<BulletinBoardMessage>> {

        public SubscriptionCallback(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback){
            super(filterList, callback);
        }

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

            // Propagate result to caller
            if (callback != null)
                callback.onSuccess(result);

        }

    }

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

        currentClient.subscribe(filterList, startEntry, new SubscriptionCallback(filterList, callback));

    }

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

}
