package forklift.consumer;

import java.util.ArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import forklift.Forklift;
import forklift.deployment.Deployment;
import forklift.deployment.DeploymentEvents;

public class ConsumerDeploymentEvents implements DeploymentEvents {
    private static final Logger log = LoggerFactory.getLogger(ConsumerDeploymentEvents.class);

    private final Map<Deployment, List<Future<?>>> running;
    private final Map<Deployment, List<ConsumerThread>> deployments;
    private final Forklift forklift;
    private final ExecutorService executor;

    public ConsumerDeploymentEvents(Forklift forklift, ExecutorService executor) {
        this.running = new HashMap<>();
        this.deployments = new HashMap<>();
        this.forklift = forklift;
        this.executor = executor;
    }

    public ConsumerDeploymentEvents(Forklift forklift) {
        // A cached thread pool creates new threads as needed, but will reuse previously
        // constructed threads when they are available. Automatically closes threads after
        // 60 seconds.
        this(forklift, Executors.newCachedThreadPool());
	}

    @Override
    public synchronized void onDeploy(Deployment deployment) {
        log.info("Deploying: " + deployment);

        final List<ConsumerThread> threads = new ArrayList<>();
        final List<Future<?>> futures = new ArrayList<>();

        deployment.getQueues().forEach(c -> {
            final ConsumerThread thread = new ConsumerThread(
                new Consumer(c, forklift.getConnector()));
            threads.add(thread);
            futures.add(executor.submit(thread));
        });

        deployment.getTopics().forEach(c -> {
            final ConsumerThread thread = new ConsumerThread(
                new Consumer(c, forklift.getConnector()));
            threads.add(thread);
            futures.add(executor.submit(thread));
        });

        deployments.put(deployment, threads);
        running.put(deployment, futures);
    }

    @Override
    public synchronized void onUndeploy(Deployment deployment) {
        log.info("Undeploying: " + deployment);

        final List<ConsumerThread> threads = deployments.remove(deployment);
        if (threads != null && !threads.isEmpty()) {
            threads.forEach(t -> {
                t.getConsumer().shutdown();
                try {
                    t.join(60000);
                } catch (Exception e) {
                }
            });
        }

        final List<Future<?>> futures = running.remove(deployment);
        if (futures != null && !futures.isEmpty()) {
            futures.forEach(f -> {
                if (!f.isCancelled()) f.cancel(true);
            });
        }
    }

}

@@ -578,68 +578,8 @@
 );%0A%0A
-    private final Map%3CDeployment, List%3CFuture%3C?%3E%3E%3E running;%0A
     
@@ -643,16 +643,16 @@
 yments;%0A
+
     priv
@@ -812,48 +812,8 @@
 ) %7B%0A
-        this.running = new HashMap%3C%3E();%0A
     
@@ -1439,67 +1439,8 @@
 %3E();
-%0A        final List%3CFuture%3C?%3E%3E futures = new ArrayList%3C%3E();
 %0A%0A  
@@ -1645,36 +1645,24 @@
             
-futures.add(
 executor.sub
@@ -1664,33 +1664,32 @@
 r.submit(thread)
-)
 ;%0A        %7D);%0A%0A 
@@ -1903,20 +1903,8 @@
     
-futures.add(
 exec
@@ -1922,17 +1922,16 @@
 (thread)
-)
 ;%0A      
@@ -1987,50 +1987,8 @@
 s);%0A
-        running.put(deployment, futures);%0A
     
@@ -2420,24 +2420,24 @@
 eption e) %7B%0A
+
             
@@ -2471,245 +2471,8 @@
    %7D
-%0A%0A        final List%3CFuture%3C?%3E%3E futures = running.remove(deployment);%0A        if (futures != null && !futures.isEmpty()) %7B%0A            futures.forEach(f -%3E %7B%0A                if (!f.isCancelled()) f.cancel(true);%0A            %7D);%0A        %7D
 %0A   

Remove running future map