package com.hubspot.baragon.watcher;

import java.util.Collection;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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);

  @Inject
  public BaragonStateWatcher(final BaragonStateFetcher stateFetcher,
                             Set<BaragonStateListener> listeners,
                             @Baragon PersistentWatcher watcher) {
    ExecutorService executor = newExecutor();

    final ListenerContainer<BaragonStateListener> listenerContainer = new ListenerContainer<>();
    for (BaragonStateListener listener : listeners) {
      listenerContainer.addListener(listener, executor);
    }

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

      @Override
      public void newEvent(Event event) {
        final Collection<BaragonServiceState> newState;

        switch (event.getType()) {
          case NODE_UPDATED:
            int version = event.getStat().getVersion();
            newState = stateFetcher.fetchState(version);
            break;
          case NODE_DELETED:
            newState = Collections.emptyList();
            break;
          default:
            LOG.warn("Unrecognized event type {}", event.getType());
            return;
        }

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

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

    watcher.start();
  }

  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;
      }
    });

  }
}

@@ -52,70 +52,136 @@
 til.
-Collection;%0Aimport java.util.Collections;%0Aimport java.util.Set
+ArrayDeque;%0Aimport java.util.Collection;%0Aimport java.util.Deque;%0Aimport java.util.Set;%0Aimport java.util.concurrent.BlockingQueue
 ;%0Aim
@@ -262,16 +262,65 @@
 cutors;%0A
+import java.util.concurrent.LinkedTransferQueue;%0A
 import j
@@ -929,16 +929,237 @@
 lass);%0A%0A
+  private final BaragonStateFetcher stateFetcher;%0A  private final ListenerContainer%3CBaragonStateListener%3E listenerContainer;%0A  private final ExecutorService executor;%0A  private final BlockingQueue%3CInteger%3E versionQueue;%0A%0A
   @Injec
@@ -1370,142 +1370,180 @@
     
-ExecutorService executor = newExecutor();%0A%0A    final ListenerContainer%3CBaragonStateListener%3E listenerContainer = new ListenerContainer
+this.stateFetcher = stateFetcher;%0A    this.listenerContainer = new ListenerContainer%3C%3E();%0A    this.executor = newExecutor();%0A    this.versionQueue = new LinkedTransferQueue
 %3C%3E()
@@ -1642,26 +1642,16 @@
 listener
-, executor
 );%0A    %7D
@@ -1782,65 +1782,8 @@
 ) %7B%0A
-        final Collection%3CBaragonServiceState%3E newState;%0A%0A
     
@@ -1914,50 +1914,209 @@
     
-newState = stateFetcher.fetchState(version
+versionQueue.add(version);%0A            executor.submit(new Runnable() %7B%0A%0A              @Override%0A              public void run() %7B%0A                updateToLatestVersion();%0A              %7D%0A            %7D
 );%0A 
@@ -2178,41 +2178,49 @@
     
-newState = Collections.emptyList(
+LOG.warn(%22Baragon state node was deleted%22
 );%0A 
@@ -2341,29 +2341,379 @@
     
+b
 re
-turn;%0A        %7D%0A%0A  
+ak;%0A        %7D%0A      %7D%0A    %7D, executor);%0A%0A    watcher.start();%0A  %7D%0A%0A  private void updateToLatestVersion() %7B%0A    Deque%3CInteger%3E versions = new ArrayDeque%3C%3E();%0A    versionQueue.drainTo(versions);%0A%0A    if (!versions.isEmpty()) %7B%0A      int latestVersion = versions.getLast();%0A      final Collection%3CBaragonServiceState%3E newState = stateFetcher.fetchState(latestVersion);%0A%0A
     
@@ -2786,34 +2786,32 @@
 d%3E() %7B%0A%0A        
-  
 @Override%0A      
@@ -2796,34 +2796,32 @@
       @Override%0A
-  
         public V
@@ -2869,26 +2869,24 @@
 %7B%0A          
-  
 listener.sta
@@ -2912,26 +2912,24 @@
 ;%0A          
-  
 return null;
@@ -2933,30 +2933,26 @@
 ll;%0A        
-  %7D%0A  
+%7D%0A
       %7D);%0A  
@@ -2953,56 +2953,14 @@
 %7D);%0A
-  
     %7D%0A
-    %7D, executor);%0A%0A    watcher.start();%0A
   %7D%0A

Make sure updates to Baragon state are synchronized and deduped