package com.hubspot.baragon.watcher;

import java.util.ArrayDeque;
import java.util.Collection;
import java.util.Deque;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ThreadFactory;

import com.google.common.base.Function;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.hubspot.baragon.models.BaragonServiceState;
import com.hubspot.ringleader.watcher.Event;
import com.hubspot.ringleader.watcher.EventListener;
import com.hubspot.ringleader.watcher.PersistentWatcher;
import org.apache.curator.framework.listen.ListenerContainer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Singleton
public class BaragonStateWatcher {
  private static final Logger LOG = LoggerFactory.getLogger(BaragonStateWatcher.class);

  private final BaragonStateFetcher stateFetcher;
  private final ListenerContainer<BaragonStateListener> listenerContainer;
  private final ExecutorService executor;
  private final BlockingQueue<Integer> versionQueue;

  @Inject
  public BaragonStateWatcher(final BaragonStateFetcher stateFetcher,
                             Set<BaragonStateListener> listeners,
                             @Baragon PersistentWatcher watcher) {
    this.stateFetcher = stateFetcher;
    this.listenerContainer = new ListenerContainer<>();
    this.executor = newExecutor();
    this.versionQueue = new LinkedTransferQueue<>();
    for (BaragonStateListener listener : listeners) {
      listenerContainer.addListener(listener);
    }

    watcher.getEventListenable().addListener(new EventListener() {

      @Override
      public void newEvent(Event event) {
        switch (event.getType()) {
          case NODE_UPDATED:
            int version = event.getStat().getVersion();
            versionQueue.add(version);
            executor.submit(new Runnable() {

              @Override
              public void run() {
                updateToLatestVersion();
              }
            });
            break;
          case NODE_DELETED:
            LOG.warn("Baragon state node was deleted");
            break;
          default:
            LOG.warn("Unrecognized event type {}", event.getType());
            break;
        }
      }
    }, executor);

    watcher.start();
  }

  private void updateToLatestVersion() {
    Deque<Integer> versions = new ArrayDeque<>();
    versionQueue.drainTo(versions);

    if (!versions.isEmpty()) {
      int latestVersion = versions.getLast();
      final Collection<BaragonServiceState> newState = stateFetcher.fetchState(latestVersion);

      listenerContainer.forEach(new Function<BaragonStateListener, Void>() {

        @Override
        public Void apply(BaragonStateListener listener) {
          listener.stateChanged(newState);
          return null;
        }
      });
    }
  }

  private ExecutorService newExecutor() {
    return Executors.newSingleThreadExecutor(new ThreadFactory() {

      @Override
      public Thread newThread(Runnable r) {
        Thread thread = Executors.defaultThreadFactory().newThread(r);
        thread.setName("BaragonStateWatcher");
        thread.setDaemon(true);
        return thread;
      }
    });

  }
}

@@ -31,16 +31,70 @@
 tcher;%0A%0A
+import java.io.Closeable;%0Aimport java.io.IOException;%0A
 import j
@@ -409,16 +409,136 @@
 ctory;%0A%0A
+import org.apache.curator.framework.listen.ListenerContainer;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A
 import c
@@ -851,127 +851,8 @@
 her;
-%0Aimport org.apache.curator.framework.listen.ListenerContainer;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0A@S
@@ -893,16 +893,37 @@
 Watcher 
+implements Closeable 
 %7B%0A  priv
@@ -2479,24 +2479,109 @@
 art();%0A  %7D%0A%0A
+  @Override%0A  public void close() throws IOException %7B%0A    executor.shutdown();%0A  %7D%0A%0A
   private vo

Implement Closeable and shut down the executor