package io.tetrapod.core;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.*;

/**
 * Manages service-wide thread-pools and dispatch of requests and sequenced behaviors
 */
public class Dispatcher {
   public static final Logger             logger   = LoggerFactory.getLogger(Dispatcher.class);

   private final ThreadPoolExecutor       threadPool;
   private final ExecutorService          sequential;
   private final ScheduledExecutorService scheduled;

   private final BlockingQueue<Runnable>  overflow = new LinkedBlockingQueue<>();

   private EventLoopGroup                 workerGroup;

   public Dispatcher() {
      this(8);
   }

   public Dispatcher(int maxThreads) {
      threadPool = new ThreadPoolExecutor(0, maxThreads, 1L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new ThreadFactory() {
         private final AtomicInteger counter = new AtomicInteger();

         @Override
         public Thread newThread(Runnable r) {
            return new Thread(r, "Dispatch-Pooled-" + counter.incrementAndGet());
         }
      });

      sequential = new ThreadPoolExecutor(0, 1, 1L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
         @Override
         public Thread newThread(Runnable r) {
            return new Thread(r, "Dispatch-Sequential");
         }
      });

      scheduled = Executors.newScheduledThreadPool(0, new ThreadFactory() {
         @Override
         public Thread newThread(Runnable r) {
            return new Thread(r, "Dispatch-Scheduled");
         }
      });
   }

   public synchronized EventLoopGroup getWorkerGroup() {
      if (workerGroup == null) {
         workerGroup = new NioEventLoopGroup();
      }
      return workerGroup;
   }

   /**
    * Dispatches a task on our thread-pool.
    */
   public void dispatch(final Runnable r) {
      assert r != null;
      try {
         threadPool.submit(new Runnable() {
            public void run() {
               processTask(r);
            }
         });
      } catch (RejectedExecutionException e) {
         overflow.add(r);
      }
   }

   /**
    * Executes a task and then drains any overflow tasks
    */
   private void processTask(final Runnable task) {
      try {
         task.run();
      } catch (Throwable e) {
         logger.error(e.getMessage(), e);
      } finally {
         processOverflow();
      }
   }

   /**
    * Drain any items we've queued in the overflow queue
    */
   private void processOverflow() {
      Runnable r = overflow.poll();
      while (r != null) {
         try {
            r.run();
         } catch (Throwable e) {
            logger.error(e.getMessage(), e);
         } finally {
            r = overflow.poll();
         }
      }
   }

   /**
    * Queues a task for sequential dispatch
    */
   public void dispatchSequential(Runnable r) {
      assert r != null;
      try {
         sequential.execute(r);
      } catch (RejectedExecutionException e) {
         overflow.add(r);
      }
   }

   /**
    * Schedule a task to be executed on the thread-pool at a later time
    */
   public ScheduledFuture<?> dispatch(int delay, TimeUnit unit, final Runnable r) {
      return scheduled.schedule(new Runnable() {
         public void run() {
            dispatch(r);
         }
      }, delay, unit);
   }

   /**
    * Cleanly terminate all thread-pools
    */
   public void shutdown() {
      threadPool.shutdown();
      sequential.shutdown();
      scheduled.shutdown();
      if (workerGroup != null) {
         workerGroup.shutdownGracefully();
      }
   }

}

@@ -1309,32 +1309,101 @@
 readFactory() %7B%0A
+         private final AtomicInteger counter = new AtomicInteger();%0A%0A
          @Overri
@@ -1505,17 +1505,46 @@
 quential
-%22
+-%22 + counter.incrementAndGet()
 );%0A     
@@ -1611,17 +1611,17 @@
 eadPool(
-0
+1
 , new Th
@@ -1628,32 +1628,101 @@
 readFactory() %7B%0A
+         private final AtomicInteger counter = new AtomicInteger();%0A%0A
          @Overri
@@ -1823,17 +1823,46 @@
 cheduled
-%22
+-%22 + counter.incrementAndGet()
 );%0A     
@@ -3902,10 +3902,87 @@
 %7D%0A   %7D%0A%0A
+   public boolean isRunning() %7B%0A      return !threadPool.isShutdown();%0A   %7D%0A%0A
 %7D%0A

Fix a weird threading bug in dispatcher