Merge pull request #27 from SWAT-engineering/improved-overflow-support/jdk-file-tree-watch

Improved overflow support: Recursive directory watches

Author: sungshik

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

@@ -68,14 +68,16 @@ public enum Kind {
     private final Path rootPath;
     private final Path relativePath;
 
+    private static final Path EMPTY_PATH = Path.of("");
+
     public WatchEvent(Kind kind, Path rootPath) {
         this(kind, rootPath, null);
     }
 
     public WatchEvent(Kind kind, Path rootPath, @Nullable Path relativePath) {
         this.kind = kind;
         this.rootPath = rootPath;
-        this.relativePath = relativePath == null ? Path.of("") : relativePath;
+        this.relativePath = relativePath == null ? EMPTY_PATH : relativePath;
     }
 
     public Kind getKind() {
@@ -105,6 +107,20 @@ public Path calculateFullPath() {
         return rootPath.resolve(relativePath);
     }
 
+    /**
+     * @return The file name of the full path of this event, or {@code null} if
+     * it has zero elements (cf. {@link Path#getFileName()}), but without
+     * calculating the full path. This method is equivalent to, but more
+     * efficient than, {@code calculateFullPath().getFileName()}.
+     */
+    public @Nullable Path getFileName() {
+        var fileName = relativePath.getFileName();
+        if (fileName == null || fileName.equals(EMPTY_PATH)) {
+            fileName = rootPath.getFileName();
+        }
+        return fileName;
+    }
+
     @Override
     public String toString() {
         return String.format("WatchEvent[%s, %s, %s]", this.rootPath, this.kind, this.relativePath);

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

@@ -34,14 +34,15 @@
 import java.util.concurrent.Executor;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
 
 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.JDKFileTreeWatch;
 import engineering.swat.watch.impl.jdk.JDKFileWatch;
-import engineering.swat.watch.impl.jdk.JDKRecursiveDirectoryWatch;
 import engineering.swat.watch.impl.overflows.IndexingRescanner;
 import engineering.swat.watch.impl.overflows.MemorylessRescanner;
 
@@ -61,6 +62,8 @@ public class Watcher {
 
     private static final BiConsumer<EventHandlingWatch, WatchEvent> EMPTY_HANDLER = (w, e) -> {};
     private volatile BiConsumer<EventHandlingWatch, WatchEvent> eventHandler = EMPTY_HANDLER;
+    private static final Predicate<WatchEvent> TRUE_FILTER = e -> true;
+    private volatile Predicate<WatchEvent> eventFilter = TRUE_FILTER;
 
     private Watcher(Path path, WatchScope scope) {
         this.path = path;
@@ -138,6 +141,22 @@ public Watcher on(WatchEventListener listener) {
         return this;
     }
 
+    /**
+     * Configures the event filter to determine which events should be passed to
+     * the event handler. By default (without calling this method), all events
+     * are passed. This method must be called at most once.
+     * @param predicate The predicate to determine an event should be kept
+     * ({@code true}) or dropped ({@code false})
+     * @return {@code this} (to support method chaining)
+     */
+    Watcher filter(Predicate<WatchEvent> predicate) {
+        if (this.eventFilter != TRUE_FILTER) {
+            throw new IllegalArgumentException("filter cannot be set more than once");
+        }
+        this.eventFilter = predicate;
+        return this;
+    }
+
     /**
      * Optionally configure the executor in which the {@link #on(Consumer)} callbacks are scheduled.
      * If not defined, every task will be scheduled on the {@link java.util.concurrent.ForkJoinPool#commonPool()}.
@@ -180,26 +199,26 @@ public ActiveWatch start() throws IOException {
 
         switch (scope) {
             case PATH_AND_CHILDREN: {
-                var result = new JDKDirectoryWatch(path, executor, h);
+                var result = new JDKDirectoryWatch(path, executor, h, eventFilter);
                 result.open();
                 return result;
             }
             case PATH_AND_ALL_DESCENDANTS: {
                 try {
-                    var result = new JDKDirectoryWatch(path, executor, eventHandler, true);
+                    var result = new JDKDirectoryWatch(path, executor, h, eventFilter, true);
                     result.open();
                     return result;
                 } catch (Throwable ex) {
                     // no native support, use the simulation
                     logger.debug("Not possible to register the native watcher, using fallback for {}", path);
                     logger.trace(ex);
-                    var result = new JDKRecursiveDirectoryWatch(path, executor, eventHandler);
+                    var result = new JDKFileTreeWatch(path, executor, h, eventFilter);
                     result.open();
                     return result;
                 }
             }
             case PATH_ONLY: {
-                var result = new JDKFileWatch(path, executor, h);
+                var result = new JDKFileWatch(path, executor, h, eventFilter);
                 result.open();
                 return result;
             }

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

@@ -32,6 +32,7 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
+import java.util.function.Predicate;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -46,12 +47,17 @@ public abstract class JDKBaseWatch implements EventHandlingWatch {
     protected final Path path;
     protected final Executor exec;
     protected final BiConsumer<EventHandlingWatch, WatchEvent> eventHandler;
+    protected final Predicate<WatchEvent> eventFilter;
     protected final AtomicBoolean started = new AtomicBoolean();
 
-    protected JDKBaseWatch(Path path, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
+    protected JDKBaseWatch(Path path, Executor exec,
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler,
+            Predicate<WatchEvent> eventFilter) {
+
         this.path = path;
         this.exec = exec;
         this.eventHandler = eventHandler;
+        this.eventFilter = eventFilter;
     }
 
     public void open() throws IOException {
@@ -99,7 +105,7 @@ protected WatchEvent translate(java.nio.file.WatchEvent<?> jdkEvent) {
         return event;
     }
 
-    private WatchEvent.Kind translate(java.nio.file.WatchEvent.Kind<?> jdkKind) {
+    protected WatchEvent.Kind translate(java.nio.file.WatchEvent.Kind<?> jdkKind) {
         if (jdkKind == StandardWatchEventKinds.ENTRY_CREATE) {
             return WatchEvent.Kind.CREATED;
         }
@@ -119,12 +125,14 @@ private WatchEvent.Kind translate(java.nio.file.WatchEvent.Kind<?> jdkKind) {
     // -- EventHandlingWatch --
 
     @Override
-    public void handleEvent(WatchEvent e) {
-        eventHandler.accept(this, e);
+    public Path getPath() {
+        return path;
     }
 
     @Override
-    public Path getPath() {
-        return path;
+    public void handleEvent(WatchEvent e) {
+        if (eventFilter.test(e)) {
+            eventHandler.accept(this, e);
+        }
     }
 }

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

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.concurrent.Executor;
 import java.util.function.BiConsumer;
+import java.util.function.Predicate;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -47,19 +48,30 @@ public class JDKDirectoryWatch extends JDKBaseWatch {
     private final Logger logger = LogManager.getLogger();
     private final boolean nativeRecursive;
     private volatile @MonotonicNonNull Closeable bundledJDKWatcher;
+    private volatile boolean closed = false;
 
     private static final BundledSubscription<SubscriptionKey, List<java.nio.file.WatchEvent<?>>>
         BUNDLED_JDK_WATCHERS = new BundledSubscription<>(JDKPoller::register);
 
-    public JDKDirectoryWatch(Path directory, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
-        this(directory, exec, eventHandler, false);
+    public JDKDirectoryWatch(Path directory, Executor exec,
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler,
+            Predicate<WatchEvent> eventFilter) {
+
+        this(directory, exec, eventHandler, eventFilter, false);
     }
 
-    public JDKDirectoryWatch(Path directory, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler, boolean nativeRecursive) {
-        super(directory, exec, eventHandler);
+    public JDKDirectoryWatch(Path directory, Executor exec,
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler,
+            Predicate<WatchEvent> eventFilter, boolean nativeRecursive) {
+
+        super(directory, exec, eventHandler, eventFilter);
         this.nativeRecursive = nativeRecursive;
     }
 
+    public boolean isClosed() {
+        return closed;
+    }
+
     private void handleJDKEvents(List<java.nio.file.WatchEvent<?>> events) {
         exec.execute(() -> {
             for (var ev : events) {
@@ -80,10 +92,18 @@ public WatchScope getScope() {
         return nativeRecursive ? WatchScope.PATH_AND_ALL_DESCENDANTS : WatchScope.PATH_AND_CHILDREN;
     }
 
+    @Override
+    public void handleEvent(WatchEvent e) {
+        if (!closed) {
+            super.handleEvent(e);
+        }
+    }
+
     @Override
     public synchronized void close() throws IOException {
-        if (bundledJDKWatcher != null) {
+        if (!closed && bundledJDKWatcher != null) {
             logger.trace("Closing watch for: {}", this.path);
+            closed = true;
             bundledJDKWatcher.close();
         }
     }