package com.jivesoftware.os.amza.service;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.jivesoftware.os.amza.api.FailedToAchieveQuorumException;
import com.jivesoftware.os.amza.api.partition.PartitionProperties;
import com.jivesoftware.os.amza.api.partition.VersionedPartitionName;
import com.jivesoftware.os.amza.api.ring.RingMember;
import com.jivesoftware.os.amza.service.ring.AmzaRingReader;
import com.jivesoftware.os.amza.service.storage.PartitionCreator;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import com.jivesoftware.os.routing.bird.health.checkers.SickThreads;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 *
 */
public class AmzaSystemReady {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private final AmzaRingStoreReader ringStoreReader;
    private final PartitionCreator partitionCreator;
    private final SickPartitions sickPartitions;

    private final AtomicBoolean tookFully = new AtomicBoolean();
    private final AtomicBoolean ready = new AtomicBoolean();
    private final Map<VersionedPartitionName, Set<RingMember>> systemTookFully = Maps.newConcurrentMap();
    private final List<Callable<Void>> onReadyCallbacks = Collections.synchronizedList(Lists.newArrayList());

    public AmzaSystemReady(AmzaRingStoreReader ringStoreReader,
        PartitionCreator partitionCreator,
        SickPartitions sickPartitions,
        SickThreads sickThreads) {

        this.ringStoreReader = ringStoreReader;
        this.partitionCreator = partitionCreator;
        this.sickPartitions = sickPartitions;

        for (VersionedPartitionName versionedPartitionName : partitionCreator.getSystemPartitions()) {
            PartitionProperties properties = partitionCreator.getProperties(versionedPartitionName.getPartitionName());
            if (properties.replicated) {
                sickPartitions.sick(versionedPartitionName, new Throwable("System partition has not yet taken fully from a quorum"));
                systemTookFully.put(versionedPartitionName, Collections.newSetFromMap(Maps.newConcurrentMap()));
            }
        }

        ExecutorService readyExecutor = Executors.newSingleThreadExecutor();
        readyExecutor.submit(() -> {
            try {
                while (!tookFully.get()) {
                    Thread.sleep(100L);
                }

                ready();
                readyExecutor.shutdown();
            } catch (Exception e) {
                LOG.error("Encountered a problem while awaiting system ready, service will be parked");
                sickThreads.sick(e);
            }
        });
    }

    public void tookFully(VersionedPartitionName versionedPartitionName, RingMember ringMember) throws Exception {
        if (ready.get() || !versionedPartitionName.getPartitionName().isSystemPartition()) {
            return;
        }

        Set<RingMember> ringMembers = systemTookFully.get(versionedPartitionName);
        if (ringMembers != null) {
            ringMembers.add(ringMember);
            checkReady();
        }
    }

    public void checkReady() throws Exception {
        int systemRingSize = ringStoreReader.getRingSize(AmzaRingReader.SYSTEM_RING);
        int quorum = systemRingSize / 2;
        boolean ready = true;
        for (Entry<VersionedPartitionName, Set<RingMember>> entry : systemTookFully.entrySet()) {
            VersionedPartitionName versionedPartitionName = entry.getKey();
            Set<RingMember> ringMembers = entry.getValue();
            if (ringMembers.size() < quorum || ringMembers.size() > systemRingSize) {
                ready = false;
                sickPartitions.sick(versionedPartitionName,
                    new Throwable("System partition has not yet taken fully from a quorum: " + ringMembers.size() + " / " + quorum));
            }
        }
        if (ready) {
            tookFully.set(true);
        }
    }

    private void ready() {
        synchronized (ready) {
            if (ready.compareAndSet(false, true)) {
                LOG.info("System is ready!");
                ready.set(true);

                ready.notifyAll();
                for (VersionedPartitionName versionedPartitionName : partitionCreator.getSystemPartitions()) {
                    sickPartitions.recovered(versionedPartitionName);
                }
                systemTookFully.clear();

                try {
                    for (Callable<Void> callback : onReadyCallbacks) {
                        callback.call();
                    }
                    onReadyCallbacks.clear();
                } catch (Exception e) {
                    LOG.error("Failed onReady callback", e);
                    ready.set(false);
                }
            }
        }
    }

    public void await(long timeoutInMillis) throws Exception {
        if (!ready.get()) {
            if (timeoutInMillis == 0) {
                throw new FailedToAchieveQuorumException("System has not reached ready state");
            }
            synchronized (ready) {
                while (!ready.get()) {
                    if (timeoutInMillis == 0) {
                        throw new FailedToAchieveQuorumException("Timed out waiting for system to reach ready state");
                    }
                    long start = System.currentTimeMillis();
                    ready.wait(timeoutInMillis);
                    timeoutInMillis = Math.max(timeoutInMillis - (start - System.currentTimeMillis()), 0);
                }
            }
        }
    }

    public void onReady(Callable<Void> callable) throws Exception {
        synchronized (ready) {
            if (ready.get()) {
                callable.call();
            } else {
                onReadyCallbacks.add(callable);
            }
        }
    }
}

@@ -115,16 +115,79 @@
 t.Maps;%0A
+import com.google.common.util.concurrent.ThreadFactoryBuilder;%0A
 import c
@@ -994,16 +994,52 @@
 cutors;%0A
+import java.util.concurrent.Future;%0A
 import j
@@ -2640,16 +2640,76 @@
 xecutor(
+new ThreadFactoryBuilder().setNameFormat(%22ready-%25d%22).build()
 );%0A     
@@ -4909,122 +4909,488 @@
     
-try %7B%0A                    for (Callable%3CVoid%3E callback : onReadyCallbacks) %7B%0A                        callback.call
+ExecutorService onReadyExecutor = Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat(%22onReadyCallback-%25d%22).build());%0A                try %7B%0A                    List%3CFuture%3C?%3E%3E futures = Lists.newArrayList();%0A                    for (Callable%3CVoid%3E callback : onReadyCallbacks) %7B%0A                        futures.add(onReadyExecutor.submit(callback));%0A                    %7D%0A                    for (Future%3C?%3E future : futures) %7B%0A                        future.get
 ();%0A
@@ -5592,24 +5592,103 @@
 set(false);%0A
+                %7D finally %7B%0A                    onReadyExecutor.shutdownNow();%0A
             

add concurrency to system ready callbacks