package com.github.msemys.esjc.subscription.manager;

import com.github.msemys.esjc.ConnectionClosedException;
import com.github.msemys.esjc.Settings;
import com.github.msemys.esjc.SubscriptionDropReason;
import com.github.msemys.esjc.operation.manager.OperationTimeoutException;
import com.github.msemys.esjc.operation.manager.RetriesLimitReachedException;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static com.github.msemys.esjc.util.Preconditions.checkNotNull;
import static java.util.stream.Stream.concat;

public class SubscriptionManager {
    private static final Logger logger = LoggerFactory.getLogger(SubscriptionManager.class);

    private final Map<UUID, SubscriptionItem> activeSubscriptions = new ConcurrentHashMap<>();
    private final Queue<SubscriptionItem> waitingSubscriptions = new ArrayDeque<>();
    private final List<SubscriptionItem> retryPendingSubscriptions = new ArrayList<>();

    private final Settings settings;

    public SubscriptionManager(Settings settings) {
        checkNotNull(settings, "settings is null");
        this.settings = settings;
    }

    public Optional<SubscriptionItem> getActiveSubscription(UUID correlationId) {
        return Optional.ofNullable(activeSubscriptions.get(correlationId));
    }

    public void cleanUp(Throwable cause) {
        if (!activeSubscriptions.isEmpty() || !waitingSubscriptions.isEmpty() || !retryPendingSubscriptions.isEmpty()) {
            ConnectionClosedException connectionClosedException = new ConnectionClosedException("Connection was closed.", cause);

            concat(activeSubscriptions.values().stream(), concat(waitingSubscriptions.stream(), retryPendingSubscriptions.stream()))
                .forEach(item -> item.operation.drop(SubscriptionDropReason.ConnectionClosed, connectionClosedException));
        }

        activeSubscriptions.clear();
        waitingSubscriptions.clear();
        retryPendingSubscriptions.clear();
    }

    public void purgeSubscribedAndDropped(ChannelId connectionId) {
        List<SubscriptionItem> subscriptionsToRemove = new ArrayList<>();

        activeSubscriptions.values().stream()
            .filter(s -> s.isSubscribed && s.connectionId.equals(connectionId))
            .forEach(s -> {
                s.operation.connectionClosed();
                subscriptionsToRemove.add(s);
            });

        subscriptionsToRemove.forEach(s -> activeSubscriptions.remove(s.correlationId));
    }

    public void checkTimeoutsAndRetry(Channel connection) {
        checkNotNull(connection, "connection is null");

        List<SubscriptionItem> retrySubscriptions = new ArrayList<>();
        List<SubscriptionItem> removeSubscriptions = new ArrayList<>();

        activeSubscriptions.values().stream()
            .filter(s -> !s.isSubscribed)
            .forEach(s -> {
                if (!s.connectionId.equals(connection.id())) {
                    retrySubscriptions.add(s);
                } else if (!s.timeout.isZero() && s.lastUpdated.isElapsed(settings.operationTimeout)) {
                    String error = String.format("Subscription never got confirmation from server. UTC now: %s, operation: %s.",
                        Instant.now(), s);

                    logger.error(error);

                    if (settings.failOnNoServerResponse) {
                        s.operation.drop(SubscriptionDropReason.SubscribingError, new OperationTimeoutException(error));
                        removeSubscriptions.add(s);
                    } else {
                        retrySubscriptions.add(s);
                    }
                }
            });

        retrySubscriptions.forEach(this::scheduleSubscriptionRetry);
        removeSubscriptions.forEach(this::removeSubscription);

        if (!retryPendingSubscriptions.isEmpty()) {
            retryPendingSubscriptions.forEach(s -> {
                s.retryCount += 1;
                startSubscription(s, connection);
            });
            retryPendingSubscriptions.clear();
        }

        while (!waitingSubscriptions.isEmpty()) {
            startSubscription(waitingSubscriptions.poll(), connection);
        }
    }

    public boolean removeSubscription(SubscriptionItem item) {
        boolean removed = activeSubscriptions.remove(item.correlationId) != null;
        logger.debug("RemoveSubscription {}, result {}.", item, removed);
        return removed;
    }

    public void scheduleSubscriptionRetry(SubscriptionItem item) {
        if (!removeSubscription(item)) {
            logger.debug("RemoveSubscription failed when trying to retry {}.", item);
            return;
        }

        if (item.maxRetries >= 0 && item.retryCount >= item.maxRetries) {
            logger.debug("RETRIES LIMIT REACHED when trying to retry {}.", item);
            item.operation.drop(SubscriptionDropReason.SubscribingError,
                new RetriesLimitReachedException(item.toString(), item.retryCount));
            return;
        }

        logger.debug("retrying subscription {}.", item);
        retryPendingSubscriptions.add(item);
    }

    public void enqueueSubscription(SubscriptionItem item) {
        waitingSubscriptions.offer(item);
    }

    public void startSubscription(SubscriptionItem item, Channel connection) {
        checkNotNull(connection, "connection is null");

        if (item.isSubscribed) {
            logger.debug("StartSubscription REMOVING due to already subscribed {}.", item);
            removeSubscription(item);
            return;
        }

        item.correlationId = UUID.randomUUID();
        item.connectionId = connection.id();
        item.lastUpdated.update();

        activeSubscriptions.put(item.correlationId, item);

        if (!item.operation.subscribe(item.correlationId, connection)) {
            logger.debug("StartSubscription REMOVING AS COULD NOT SUBSCRIBE {}.", item);
            removeSubscription(item);
        } else {
            logger.debug("StartSubscription SUBSCRIBING {}.", item);
        }
    }

}

@@ -347,24 +347,25 @@
 dException;%0A
+%0A
 import io.ne
@@ -420,16 +420,17 @@
 nnelId;%0A
+%0A
 import o
@@ -571,16 +571,67 @@
 HashMap;
+%0Aimport java.util.concurrent.ConcurrentLinkedQueue;
 %0A%0Aimport
@@ -694,16 +694,17 @@
 otNull;%0A
+%0A
 import s
@@ -1039,16 +1039,27 @@
 new 
-ArrayDeq
+ConcurrentLinkedQue
 ue%3C%3E
@@ -1080,20 +1080,21 @@
 e final 
-List
+Queue
 %3CSubscri
@@ -1132,33 +1132,45 @@
 tions = new 
-ArrayList
+ConcurrentLinkedQueue
 %3C%3E();%0A%0A    p
@@ -4023,63 +4023,53 @@
     
-if (!retryPendingSubscriptions.isEmpty()) %7B%0A           
+SubscriptionItem item;%0A        while ((item =
  ret
@@ -4091,28 +4091,32 @@
 iptions.
-forEach(s -%3E
+poll()) != null)
  %7B%0A     
@@ -4122,21 +4122,20 @@
         
-    s
+item
 .retryCo
@@ -4144,20 +4144,16 @@
 t += 1;%0A
-    
         
@@ -4174,17 +4174,20 @@
 ription(
-s
+item
 , connec
@@ -4197,71 +4197,8 @@
 n);%0A
-            %7D);%0A            retryPendingSubscriptions.clear();%0A
     
@@ -4219,17 +4219,24 @@
  while (
-!
+(item = 
 waitingS
@@ -4244,33 +4244,39 @@
 bscriptions.
-isEmpty()
+poll()) != null
 ) %7B%0A        
@@ -4301,35 +4301,12 @@
 ion(
-waitingSubscriptions.poll()
+item
 , co
@@ -6193,11 +6193,10 @@
 %7D%0A    %7D%0A
-%0A
 %7D%0A

Use concurrent primitives