Improved overflow support: Recursive directory watches

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

@@ -40,6 +40,7 @@
 
 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;
@@ -186,14 +187,15 @@ public ActiveWatch start() throws IOException {
             }
             case PATH_AND_ALL_DESCENDANTS: {
                 try {
-                    var result = new JDKDirectoryWatch(path, executor, eventHandler, true);
+                    var result = new JDKDirectoryWatch(path, executor, h, 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 JDKRecursiveDirectoryWatch(path, executor, h);
+                    var result = new JDKFileTreeWatch(path, executor, h);
                     result.open();
                     return result;
                 }

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

@@ -99,7 +99,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;
         }

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

@@ -0,0 +1,268 @@
+/*
+ * BSD 2-Clause License
+ *
+ * Copyright (c) 2023, Swat.engineering
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice, this
+ *    list of conditions and the following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+ * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package engineering.swat.watch.impl.jdk;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import engineering.swat.watch.WatchEvent;
+import engineering.swat.watch.WatchScope;
+import engineering.swat.watch.impl.EventHandlingWatch;
+
+public class JDKFileTreeWatch extends JDKBaseWatch {
+    private final Logger logger = LogManager.getLogger();
+    private final Path rootPath;
+    private final Path relativePathParent;
+    private final Map<Path, JDKFileTreeWatch> childWatches = new ConcurrentHashMap<>();
+    private final JDKBaseWatch internal;
+
+    public JDKFileTreeWatch(Path fullPath, Executor exec,
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
+        this(fullPath, Path.of(""), exec, eventHandler);
+    }
+
+    public JDKFileTreeWatch(Path rootPath, Path relativePathParent, Executor exec,
+            BiConsumer<EventHandlingWatch, WatchEvent> eventHandler) {
+
+        super(rootPath.resolve(relativePathParent), exec, eventHandler);
+        this.rootPath = rootPath;
+        this.relativePathParent = relativePathParent;
+
+        var internalEventHandler = eventHandler.andThen(new AsyncChildWatchesUpdater());
+        this.internal = new JDKDirectoryWatch(path, exec, internalEventHandler) {
+
+            // 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));
+            }
+
+            // Override to ensure that this watch translates JDK events using
+            // `rootPath` (instead of `path`, as is the default behavior).
+            // Events returned by this method do not need to be relativized.
+            @Override
+            protected WatchEvent translate(java.nio.file.WatchEvent<?> jdkEvent) {
+                var kind = translate(jdkEvent.kind());
+
+                Path relativePath = null;
+                if (kind != WatchEvent.Kind.OVERFLOW) {
+                    var child = (Path) jdkEvent.context();
+                    if (child != null) {
+                        relativePath = relativePathParent.resolve(child);
+                    }
+                }
+
+                var event = new WatchEvent(kind, rootPath, relativePath);
+                logger.trace("Translated: {} to {}", jdkEvent, event);
+                return event;
+            }
+        };
+    }
+
+    /**
+     * Event handler that asynchronously (using {@link JDKBaseWatch#exec})
+     * updates the child watches according to the following rules: (a) when an
+     * overflow happens, the directory is rescanned, new child watches for
+     * created subdirectories are opened, existing child watches for deleted
+     * subdirectories are closed, and the overflow is propagated to each child
+     * watch; (b) when a subdirectory creation happens, a new child watch is
+     * opened for that subdirectory; (c) when a subdirectory deletion happens,
+     * an existing child watch is closed for that subdirectory.
+     */
+    private class AsyncChildWatchesUpdater implements BiConsumer<EventHandlingWatch, WatchEvent> {
+        @Override
+        public void accept(EventHandlingWatch watch, WatchEvent event) {
+            exec.execute(() -> {
+                switch (event.getKind()) {
+                    case OVERFLOW: acceptOverflow(); break;
+                    case CREATED: getFileNameAndThen(event, this::acceptCreated); break;
+                    case DELETED: getFileNameAndThen(event, this::acceptDeleted); break;
+                    case MODIFIED: break;
+                }
+            });
+        }
+
+        private void getFileNameAndThen(WatchEvent event, Consumer<Path> consumer) {
+            var child = event.getFileName();
+            if (child != null) {
+                consumer.accept(child);
+            } else {
+                logger.error("Could not get file name of event: {}", event);
+            }
+        }
+
+        private void acceptOverflow() {
+            openAndCloseChildWatches();
+            for (var childWatch : childWatches.values()) {
+                reportOverflowTo(childWatch);
+            }
+        }
+
+        private void acceptCreated(Path child) {
+            if (Files.isDirectory(path.resolve(child))) {
+                var childWatch = openChildWatch(child);
+                // Events in the newly created directory might have been missed
+                // between its creation and setting up its watch. So, generate
+                // an `OVERFLOW` event for the watch.
+                reportOverflowTo(childWatch);
+            }
+        }
+
+        private void acceptDeleted(Path child) {
+            tryCloseChildWatch(child);
+        }
+
+        private void reportOverflowTo(JDKFileTreeWatch childWatch) {
+            var overflow = new WatchEvent(WatchEvent.Kind.OVERFLOW,
+                childWatch.rootPath, childWatch.relativePathParent);
+            childWatch.handleEvent(overflow);
+        }
+    }
+
+    private void openAndCloseChildWatches() {
+        var toBeClosed = new HashSet<>(childWatches.keySet());
+
+        try (var children = Files.find(path, 1, (p, attrs) -> p != path && attrs.isDirectory())) {
+            children.forEach(p -> {
+                var child = p.getFileName();
+                if (child != null) {
+                    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);
+        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;
+    }
+
+    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);
+        }
+    }
+
+    private void closeChildWatch(Path child) throws IOException {
+        assert !child.isAbsolute();
+
+        var childWatch = childWatches.remove(child);
+        if (childWatch != null) {
+            childWatch.close();
+        }
+    }
+
+    // -- JDKBaseWatch --
+
+    @Override
+    public WatchScope getScope() {
+        return WatchScope.PATH_AND_ALL_DESCENDANTS;
+    }
+
+    @Override
+    public void handleEvent(WatchEvent event) {
+        internal.handleEvent(event);
+    }
+
+    @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;
+            }
+        } while (children.hasNext());
+
+        if (firstFail != null) {
+            throw firstFail;
+        }
+    }
+
+    @Override
+    protected synchronized void start() throws IOException {
+        internal.open();
+        openAndCloseChildWatches();
+        // There's no need to report an overflow event, because `internal` was
+        // opened *before* the file system was accessed to fetch children. Thus,
+        // if a new directory is created while this method is running, then at
+        // least one of the following is true: (a) the new directory is already
+        // visible by the time the file system is accessed; (b) its `CREATED`
+        // event is handled later, which starts a new child watch if needed.
+    }
+}