Merge pull request #21 from SWAT-engineering/improved-overflow-support/event-handlers-with-active-watch

Improved overflow support: Internal event handlers can dispatch new events

Author: sungshik

src/main/java/engineering/swat/watch/Watcher.java

@@ -32,11 +32,13 @@
 import java.nio.file.Path;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import engineering.swat.watch.impl.EventHandlingWatch;
 import engineering.swat.watch.impl.jdk.JDKDirectoryWatch;
 import engineering.swat.watch.impl.jdk.JDKFileWatch;
 import engineering.swat.watch.impl.jdk.JDKRecursiveDirectoryWatch;
@@ -54,8 +56,8 @@ public class Watcher {
     private final Path path;
     private volatile Executor executor = CompletableFuture::runAsync;
 
-    private static final Consumer<WatchEvent> EMPTY_HANDLER = p -> {};
-    private volatile Consumer<WatchEvent> eventHandler = EMPTY_HANDLER;
+    private static final BiConsumer<EventHandlingWatch, WatchEvent> EMPTY_HANDLER = (w, e) -> {};
+    private volatile BiConsumer<EventHandlingWatch, WatchEvent> eventHandler = EMPTY_HANDLER;
 
 
     private Watcher(WatchScope scope, Path path) {
@@ -103,7 +105,7 @@ public Watcher on(Consumer<WatchEvent> eventHandler) {
         if (this.eventHandler != EMPTY_HANDLER) {
             throw new IllegalArgumentException("on handler cannot be set more than once");
         }
-        this.eventHandler = eventHandler;
+        this.eventHandler = (w, e) -> eventHandler.accept(e);
         return this;
     }
 
@@ -114,7 +116,7 @@ public Watcher on(WatchEventListener listener) {
         if (this.eventHandler != EMPTY_HANDLER) {
             throw new IllegalArgumentException("on handler cannot be set more than once");
         }
-        this.eventHandler = ev -> {
+        this.eventHandler = (w, ev) -> {
             switch (ev.getKind()) {
                 case CREATED:
                     listener.onCreated(ev);

src/main/java/engineering/swat/watch/impl/jdk/JDKBaseWatch.java

@@ -31,7 +31,7 @@
 import java.nio.file.StandardWatchEventKinds;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -45,10 +45,10 @@ public abstract class JDKBaseWatch implements EventHandlingWatch {
 
     protected final Path path;
     protected final Executor exec;
-    protected final Consumer<WatchEvent> eventHandler;
+    protected final BiConsumer<EventHandlingWatch, WatchEvent> eventHandler;
     protected final AtomicBoolean started = new AtomicBoolean();
 
-    protected JDKBaseWatch(Path path, Executor exec, Consumer<WatchEvent> eventHandler) {
+    protected JDKBaseWatch(Path path, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
         this.path = path;
         this.exec = exec;
         this.eventHandler = eventHandler;
@@ -120,7 +120,7 @@ private WatchEvent.Kind translate(java.nio.file.WatchEvent.Kind<?> jdkKind) {
 
     @Override
     public void handleEvent(WatchEvent e) {
-        eventHandler.accept(e);
+        eventHandler.accept(this, e);
     }
 
     @Override

src/main/java/engineering/swat/watch/impl/jdk/JDKDirectoryWatch.java

@@ -31,13 +31,14 @@
 import java.nio.file.Path;
 import java.util.List;
 import java.util.concurrent.Executor;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 
 import engineering.swat.watch.WatchEvent;
+import engineering.swat.watch.impl.EventHandlingWatch;
 import engineering.swat.watch.impl.util.BundledSubscription;
 import engineering.swat.watch.impl.util.SubscriptionKey;
 
@@ -49,11 +50,11 @@ public class JDKDirectoryWatch extends JDKBaseWatch {
     private static final BundledSubscription<SubscriptionKey, List<java.nio.file.WatchEvent<?>>>
         BUNDLED_JDK_WATCHERS = new BundledSubscription<>(JDKPoller::register);
 
-    public JDKDirectoryWatch(Path directory, Executor exec, Consumer<WatchEvent> eventHandler) {
+    public JDKDirectoryWatch(Path directory, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
         this(directory, exec, eventHandler, false);
     }
 
-    public JDKDirectoryWatch(Path directory, Executor exec, Consumer<WatchEvent> eventHandler, boolean nativeRecursive) {
+    public JDKDirectoryWatch(Path directory, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler, boolean nativeRecursive) {
         super(directory, exec, eventHandler);
         this.nativeRecursive = nativeRecursive;
     }

src/main/java/engineering/swat/watch/impl/jdk/JDKFileWatch.java

@@ -29,13 +29,14 @@
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.concurrent.Executor;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import engineering.swat.watch.WatchEvent;
+import engineering.swat.watch.impl.EventHandlingWatch;
 
 /**
  * It's not possible to monitor a single file (or directory), so we have to find a directory watcher, and connect to that
@@ -46,17 +47,17 @@ public class JDKFileWatch extends JDKBaseWatch {
     private final Logger logger = LogManager.getLogger();
     private final JDKBaseWatch internal;
 
-    public JDKFileWatch(Path file, Executor exec, Consumer<WatchEvent> eventHandler) {
+    public JDKFileWatch(Path file, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
         super(file, exec, eventHandler);
 
         var message = "The root path is not a valid path for a file watch";
         var parent = requireNonNull(file.getParent(), message);
         var fileName = requireNonNull(file.getFileName(), message);
         assert !parent.equals(file);
 
-        this.internal = new JDKDirectoryWatch(parent, exec, e -> {
+        this.internal = new JDKDirectoryWatch(parent, exec, (w, e) -> {
             if (fileName.equals(e.getRelativePath())) {
-                eventHandler.accept(e);
+                eventHandler.accept(w, e);
             }
         });
 

src/main/java/engineering/swat/watch/impl/jdk/JDKRecursiveDirectoryWatch.java

@@ -42,24 +42,25 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import engineering.swat.watch.WatchEvent;
+import engineering.swat.watch.impl.EventHandlingWatch;
 
 public class JDKRecursiveDirectoryWatch extends JDKBaseWatch {
     private final Logger logger = LogManager.getLogger();
     private final ConcurrentMap<Path, JDKDirectoryWatch> activeWatches = new ConcurrentHashMap<>();
 
-    public JDKRecursiveDirectoryWatch(Path directory, Executor exec, Consumer<WatchEvent> eventHandler) {
+    public JDKRecursiveDirectoryWatch(Path directory, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
         super(directory, exec, eventHandler);
     }
 
     private void processEvents(WatchEvent ev) {
         logger.trace("Forwarding event: {}", ev);
-        eventHandler.accept(ev);
+        eventHandler.accept(this, ev);
         logger.trace("Unwrapping event: {}", ev);
         switch (ev.getKind()) {
             case CREATED: handleCreate(ev); break;
@@ -71,10 +72,9 @@ private void processEvents(WatchEvent ev) {
 
     private void publishExtraEvents(List<WatchEvent> ev) {
         logger.trace("Reporting new nested directories & files: {}", ev);
-        ev.forEach(eventHandler);
+        ev.forEach(e -> eventHandler.accept(this, e));
     }
 
-
     private void handleCreate(WatchEvent ev) {
         // between the event and the current state of the file system
         // we might have some nested directories we missed
@@ -161,9 +161,9 @@ private void addNewDirectory(Path dir) throws IOException {
         }
 
         /** Make sure that the events are relative to the actual root of the recursive watch */
-        private Consumer<WatchEvent> relocater(Path subRoot) {
+        private BiConsumer<EventHandlingWatch, WatchEvent> relocater(Path subRoot) {
             final Path newRelative = path.relativize(subRoot);
-            return ev -> {
+            return (w, ev) -> {
                 var rewritten = new WatchEvent(ev.getKind(), path, newRelative.resolve(ev.getRelativePath()));
                 processEvents(rewritten);
             };