Remove JDK watch only after a delay, just to avoid hammering the registry api

Author: DavyLandman

src/main/java/engineering/swat/watch/impl/BundledSubscription.java

@@ -3,11 +3,15 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
@@ -16,6 +20,7 @@
  * This is used (for example) to avoid multiple JDKPoller registries for the same path
  */
 public class BundledSubscription<Key extends @NonNull Object, Event extends @NonNull Object> implements ISubscribable<Key,Event> {
+    private static final Logger logger = LogManager.getLogger();
     private final ISubscribable<Key, Event> wrapped;
     private final ConcurrentMap<Key, Subscription<Event>> subscriptions = new ConcurrentHashMap<>();
 
@@ -71,15 +76,33 @@ public Closeable subscribe(Key target, Consumer<Event> eventListener) throws IOE
                 }
             }
             return () -> {
+                boolean scheduleClose = false;
                 synchronized(active) {
                     active.remove(eventListener);
-                    if (!active.hasActiveConsumers() && !active.closed) {
-                        active.closed = true;
-                        this.subscriptions.remove(target, active);
-                        if (active.toBeClosed != null) {
-                            active.toBeClosed.close();
-                        }
-                    }
+                    scheduleClose = !active.hasActiveConsumers() && !active.closed;
+                }
+                if (scheduleClose) {
+                    // to avoid hammering the system with closes & registers in a short periode
+                    // we schedule the cleanup of watches in the background, when even after a small delay
+                    // nobody is interested in a certain file anymore
+                    CompletableFuture
+                        .delayedExecutor(100, TimeUnit.MILLISECONDS)
+                        .execute(() -> {
+                            synchronized(active) {
+                                if (!active.hasActiveConsumers() && !active.closed) {
+                                    // still ready to be closed
+                                    active.closed = true;
+                                    this.subscriptions.remove(target, active);
+                                    if (active.toBeClosed != null) {
+                                        try {
+                                            active.toBeClosed.close();
+                                        } catch (IOException e) {
+                                            logger.error("Unhandled exception while closing the watcher for {} in the background", target, e);
+                                        }
+                                    }
+                                }
+                            }
+                        });
                 }
             };
         }