Improved overflow support: Recursive directory watches

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

@@ -34,6 +34,7 @@
 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;
@@ -42,7 +43,6 @@
 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;
 
@@ -62,6 +62,8 @@
 
     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;
@@ -139,6 +141,22 @@
         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
+     * (`true`) or dropped (`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()}.
@@ -181,27 +199,26 @@
 
         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, h, 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, h);
-                    var result = new JDKFileTreeWatch(path, executor, h);
+                    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 {
@@ -118,11 +124,6 @@ protected WatchEvent.Kind translate(java.nio.file.WatchEvent.Kind<?> jdkKind) {
 
     // -- EventHandlingWatch --
 
-    @Override
-    public void handleEvent(WatchEvent e) {
-        eventHandler.accept(this, e);
-    }
-
     @Override
     public Path getPath() {
         return path;

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,27 +48,36 @@
     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;
     }
 
     private void handleJDKEvents(List<java.nio.file.WatchEvent<?>> events) {
         exec.execute(() -> {
-            for (var ev : events) {
-                try {
-                    handleEvent(translate(ev));
-                }
-                catch (Throwable ignored) {
-                    logger.error("Ignoring downstream exception:", ignored);
+            if (!closed) {
+                for (var ev : events) {
+                    try {
+                        handleEvent(translate(ev));
+                    }
+                    catch (Throwable ignored) {
+                        logger.error("Ignoring downstream exception:", ignored);
+                    }
                 }
             }
         });
@@ -80,10 +90,18 @@
         return nativeRecursive ? WatchScope.PATH_AND_ALL_DESCENDANTS : WatchScope.PATH_AND_CHILDREN;
     }
 
+    @Override
+    public void handleEvent(WatchEvent e) {
+        if (eventFilter.test(e)) {
+            eventHandler.accept(this, 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();
         }
     }

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

@@ -26,6 +26,7 @@
  */
 package engineering.swat.watch.impl.jdk;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -36,9 +37,11 @@
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 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.WatchScope;
@@ -52,27 +55,39 @@
     private final JDKBaseWatch internal;
 
     public JDKFileTreeWatch(Path fullPath, Executor exec,
-            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
-        this(fullPath, Path.of(""), exec, eventHandler);
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler,
+            Predicate<WatchEvent> eventFilter) {
+
+        this(fullPath, Path.of(""), exec, eventHandler, eventFilter);
     }
 
     public JDKFileTreeWatch(Path rootPath, Path relativePathParent, Executor exec,
-            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler,
+            Predicate<WatchEvent> eventFilter) {
 
-        super(rootPath.resolve(relativePathParent), exec, eventHandler);
+        super(rootPath.resolve(relativePathParent), exec, eventHandler, eventFilter);
         this.rootPath = rootPath;
         this.relativePathParent = relativePathParent;
 
         var internalEventHandler = eventHandler.andThen(new AsyncChildWatchesUpdater());
-        this.internal = new JDKDirectoryWatch(path, exec, internalEventHandler) {
+        this.internal = new JDKDirectoryWatch(path, exec, internalEventHandler, eventFilter) {
 
             // Override to ensure that this watch relativizes events wrt
             // `rootPath` (instead of `path`, as is the default behavior)
             @Override
             public WatchEvent relativize(WatchEvent event) {
-                var fileName = event.getFileName();
-                return new WatchEvent(event.getKind(), rootPath,
-                    fileName == null ? relativePathParent : relativePathParent.resolve(fileName));
+                var relativePath = relativePathParent;
+
+                // Append a file name to `relativePath` if it exists
+                var fullPath = event.calculateFullPath();
+                if (!fullPath.equals(path)) {
+                    var fileName = fullPath.getFileName();
+                    if (fileName != null) {
+                        relativePath = relativePath.resolve(fileName);
+                    }
+                }
+
+                return new WatchEvent(event.getKind(), rootPath, relativePath);
             }
 
             // Override to ensure that this watch translates JDK events using
@@ -125,7 +140,7 @@
             if (child != null) {
                 consumer.accept(child);
             } else {
                 logger.error("Could not get file name of event: {}", event);
             }
         }
 
@@ -167,28 +182,28 @@
                     toBeClosed.remove(child);
                     openChildWatch(child);
                 } else {
                     logger.error("File tree watch (for: {}) could not open a child watch for: {}", path, p);
                 }
             });
         } catch (IOException e) {
             logger.error("File tree watch (for: {}) could not iterate over its children ({})", path, e);
         }
 
         for (var child : toBeClosed) {
             tryCloseChildWatch(child);
         }
     }
 
     private JDKFileTreeWatch openChildWatch(Path child) {
         assert !child.isAbsolute();
 
         Function<Path, JDKFileTreeWatch> newChildWatch = p -> new JDKFileTreeWatch(
-            rootPath, relativePathParent.resolve(child), exec, eventHandler);
+            rootPath, relativePathParent.resolve(child), exec, eventHandler, eventFilter);
         var childWatch = childWatches.computeIfAbsent(child, newChildWatch);
         try {
             childWatch.startIfFirstTime();
         } catch (IOException e) {
             logger.error("Could not open (nested) file tree watch for: {} ({})", child, e);
         }
         return childWatch;
     }
@@ -196,8 +211,8 @@
     private void tryCloseChildWatch(Path child) {
         try {
             closeChildWatch(child);
         } catch (IOException e) {
             logger.error("Could not close (nested) file tree watch for: {} ({})", path.resolve(child), e);
         }
     }
 
@@ -206,15 +221,28 @@
 
         var childWatch = childWatches.remove(child);
         if (childWatch != null) {
             childWatch.close();
         }
     }
 
+    private @Nullable IOException tryClose(Closeable c) {
+        try {
+            c.close();
+            return null;
+        } catch (IOException ex) {
+            logger.error("Could not close watch", ex);
+            return ex;
+        } catch (Exception ex) {
+            logger.error("Could not close watch", ex);
+            return new IOException("Unexpected exception when closing", ex);
+        }
+    }
+
     // -- JDKBaseWatch --
 
     @Override
     public WatchScope getScope() {
         return WatchScope.PATH_AND_ALL_DESCENDANTS;
     }
 
     @Override
@@ -224,37 +252,19 @@
 
     @Override
     public synchronized void close() throws IOException {
-        IOException firstFail = null;
-
-        var internalOpen = true;
-        var children = childWatches.keySet().iterator();
-        do {
-            try {
-                // First, close the internal watch to prevent new child watches
-                // from being opened concurrently while this method is running.
-                if (internalOpen) {
-                    internal.close();
-                    internalOpen = false;
-                }
-                // Next, close all child watches
-                else {
-                    closeChildWatch(children.next());
-                }
-            } catch (IOException ex) {
-                logger.error("Could not close watch", ex);
-                firstFail = firstFail == null ? ex : firstFail;
-            } catch (Exception ex) {
-                logger.error("Could not close watch", ex);
-                firstFail = firstFail == null ? new IOException("Unexpected exception when closing", ex) : firstFail;
+        var firstFail = tryClose(internal);
+        for (var c : childWatches.values()) {
+            var currentFail = tryClose(c);
+            if (currentFail != null && firstFail == null) {
+                firstFail = currentFail;
             }
-        } while (children.hasNext());
-
+        }
         if (firstFail != null) {
             throw firstFail;
         }
     }
 
-    @Override
+   @Override
     protected synchronized void start() throws IOException {
         internal.open();
         openAndCloseChildWatches();

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

@@ -30,6 +30,7 @@
 import java.nio.file.Path;
 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;
@@ -48,8 +49,11 @@ public class JDKFileWatch extends JDKBaseWatch {
     private final Logger logger = LogManager.getLogger();
     private final JDKBaseWatch internal;
 
-    public JDKFileWatch(Path file, Executor exec, BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
-        super(file, exec, eventHandler);
+    public JDKFileWatch(Path file, Executor exec,
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler,
+            Predicate<WatchEvent> eventFilter) {
+
+        super(file, exec, eventHandler, eventFilter);
 
         var message = "The root path is not a valid path for a file watch";
         var parent = requireNonNull(file.getParent(), message);
@@ -64,7 +68,7 @@ public JDKFileWatch(Path file, Executor exec, BiConsumer<EventHandlingWatch, Wat
             if (fileName.equals(e.getRelativePath())) {
                 eventHandler.accept(w, e);
             }
-        });
+        }, eventFilter);
 
         logger.debug("File watch (for: {}) is in reality a directory watch (for: {}) with a filter (for: {})", file, parent, fileName);
     }