1 files changed, 52 insertions, 10 deletions

diff --git a/src/main/java/io/trygvis/esper/testing/ResourceManager.java b/src/main/java/io/trygvis/esper/testing/ResourceManager.java
index e9a0068..dff19bd 100644
--- a/src/main/java/io/trygvis/esper/testing/ResourceManager.java
+++ b/src/main/java/io/trygvis/esper/testing/ResourceManager.java
@@ -1,31 +1,73 @@
 package io.trygvis.esper.testing;
 
-import fj.*;
-
+import java.io.*;
 import java.util.*;
 import java.util.concurrent.*;
 
-public class ResourceManager<K, V> {
-    private final Equal<K> equal;
-    private final Callable<List<K>> discoverer;
+public class ResourceManager<K extends Comparable<K>, V> implements Closeable {
+    private final ResourceManagerCallbacks<K, V> callbacks;
+    private final ScheduledFuture<?> future;
+
     private Map<K, V> map = Collections.emptyMap();
 
-    public ResourceManager(Equal<K> equal, ScheduledExecutorService executorService, int delay, Callable<List<K>> discoverer) {
-        this.equal = equal;
-        this.discoverer = discoverer;
+    public interface ResourceManagerCallbacks<K extends Comparable<K>, V> {
+        Set<K> discover() throws Exception;
+
+        V onNew(K key);
+
+        void onGone(K key, V value);
+    }
+
+    public ResourceManager(ScheduledExecutorService executorService, int delay,
+                           ResourceManagerCallbacks<K, V> callbacks) {
+        this.callbacks = callbacks;
 
-        executorService.scheduleWithFixedDelay(new Runnable() {
+        future = executorService.scheduleWithFixedDelay(new Runnable() {
             public void run() {
                 work();
             }
         }, delay, delay, TimeUnit.MILLISECONDS);
+
     }
 
     private void work() {
         try {
-            List<K> keys = discoverer.call();
+            System.out.println("Discovering...");
+            Set<K> keys = callbacks.discover();
+
+            Set<K> found = new HashSet<>();
+
+            for (K key : keys) {
+                if (map.containsKey(key)) {
+                    continue;
+                }
+
+                found.add(key);
+            }
+
+            Set<K> lost = new HashSet<>(map.keySet());
+            lost.retainAll(found);
+
+            System.out.println("Discovered " + keys.size() + " keys, new: " + found.size() + ", gone=" + lost.size());
+
+            Map<K, V> newMap = new HashMap<>(found.size());
+
+            for (K k : found) {
+                newMap.put(k, callbacks.onNew(k));
+            }
+
+            for (K k : lost) {
+                callbacks.onGone(k, map.get(k));
+            }
+
+            map = newMap;
         } catch (Exception e) {
+            e.printStackTrace();
             return;
         }
     }
+
+    public void close() {
+        future.cancel(true);
+    }
 }