Merge pull request #20 from SWAT-engineering/improved-overflow-support-main

Improved overflow support

Author: DavyLandman

.github/workflows/build.yaml

@@ -8,6 +8,7 @@ on:
   pull_request:
     branches:
       - main
+      - improved-overflow-support-main
 
 jobs:
   test:

README.md

@@ -7,13 +7,14 @@ a java file watcher that works across platforms and supports recursion, single f
 Features:
 
 - monitor a single file (or directory) for changes
-- monitor a directory for changes to it's direct descendants
-- monitor a directory for changes for all it's descendants (aka recursive directory watch)
+- monitor a directory for changes to its direct descendants
+- monitor a directory for changes for all its descendants (aka recursive directory watch)
 - edge cases dealt with:
-  - in case of overflow we will still generate events for new descendants
   - recursive watches will also continue in new directories
   - multiple watches for the same directory are merged to avoid overloading the kernel
   - events are processed in a configurable worker pool
+  - when an overflow happens, automatically approximate the events that were
+    missed using a configurable approximation policy
 
 Planned features:
 
@@ -41,6 +42,7 @@ Start using java-watch:
 var directory = Path.of("tmp", "test-dir");
 var watcherSetup = Watcher.watch(directory, WatchScope.PATH_AND_CHILDREN)
     .withExecutor(Executors.newCachedThreadPool()) // optionally configure a custom thread pool
+    .onOverflow(Approximation.DIFF) // optionally configure a handler for overflows
     .on(watchEvent -> {
         System.err.println(watchEvent);
     });

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

@@ -27,12 +27,22 @@
 package engineering.swat.watch;
 
 import java.io.Closeable;
+import java.nio.file.Path;
 
 /**
- * <p>Marker interface for an active watch, in the future might get properties you can inspect.</p>
+ * <p>Marker interface for an active watch, in the future might get more properties you can inspect.</p>
  *
- * <p>For now, make sure to close the watch when not interested in new events</p>
+ * <p>For now, make sure to close the watch when not interested in new events.</p>
  */
 public interface ActiveWatch extends Closeable {
 
+    /**
+     * Gets the path watched by this watch.
+     */
+    Path getPath();
+
+    /**
+     * Gets the scope of this watch.
+     */
+    WatchScope getScope();
 }

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

@@ -0,0 +1,108 @@
+/*
+ * 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;
+
+/**
+ * Constants to indicate for which regular files/directories in the scope of the
+ * watch an <i>approximation</i> of synthetic events (of kinds
+ * {@link WatchEvent.Kind#CREATED}, {@link WatchEvent.Kind#MODIFIED}, and/or
+ * {@link WatchEvent.Kind#DELETED}) should be issued when an overflow event
+ * happens. These synthetic events, as well as the overflow event itself, are
+ * subsequently passed to the user-defined event handler of the watch.
+ * Typically, the user-defined event handler can ignore the original overflow
+ * event (i.e., handling the synthetic events is sufficient to address the
+ * overflow issue), but it doesn't have to (e.g., it may carry out additional
+ * overflow bookkeeping).
+ */
+public enum Approximation {
+
+    /**
+     * Synthetic events are issued for <b>no regular files/directories</b> in
+     * the scope of the watch. Thus, the user-defined event handler is fully
+     * responsible to handle overflow events.
+     */
+    NONE,
+
+    /**
+     * <p>
+     * Synthetic events of kinds {@link WatchEvent.Kind#CREATED} and
+     * {@link WatchEvent.Kind#MODIFIED}, but not
+     * {@link WatchEvent.Kind#DELETED}, are issued for all regular
+     * files/directories in the scope of the watch. Specifically, when an
+     * overflow event happens:
+     *
+     * <ul>
+     * <li>CREATED events are issued for all regular files/directories
+     * (overapproximation).
+     * <li>MODIFIED events are issued for all non-empty, regular files
+     * (overapproximation) but for no directories (underapproximation).
+     * <li>DELETED events are issued for no regular files/directories
+     * (underapproximation).
+     * </ul>
+     *
+     * <p>
+     * This approach is relatively cheap in terms of memory usage (cf.
+     * {@link #DIFF}), but it results in a large over/underapproximation of the
+     * actual events (cf. DIFF).
+     */
+    ALL,
+
+
+    /**
+     * <p>
+     * Synthetic events of kinds {@link WatchEvent.Kind#CREATED},
+     * {@link WatchEvent.Kind#MODIFIED}, and {@link WatchEvent.Kind#DELETED} are
+     * issued for regular files/directories in the scope of the watch, when
+     * their current versions are different from their previous versions, as
+     * determined using <i>last-modified-times</i>. Specifically, when an
+     * overflow event happens:
+     *
+     * <ul>
+     * <li>CREATED events are issued for all regular files/directories when the
+     * previous last-modified-time is unknown, but the current
+     * last-modified-time is known (i.e., the file started existing).
+     * <li>MODIFIED events are issued for all regular files/directories when the
+     * previous last-modified-time is before the current last-modified-time.
+     * <li>DELETED events are issued for all regular files/directories when the
+     * previous last-modified-time is known, but the current
+     * last-modified-time is unknown (i.e., the file stopped existing).
+     * </ul>
+     *
+     * <p>
+     * To keep track of last-modified-times, an internal <i>index</i> is
+     * populated with last-modified-times of all regular files/directories in
+     * the scope of the watch when the watch is started. Each time when any
+     * event happens, the index is updated accordingly, so when an overflow
+     * event happens, last-modified-times can be compared as described above.
+     *
+     * <p>
+     * This approach results in a small overapproximation (cf. {@link #ALL}),
+     * but it is relatively expensive in terms of memory usage (cf. ALL), as the
+     * watch needs to keep track of last-modified-times.
+     */
+    DIFF
+}

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

@@ -68,10 +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() {
@@ -101,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

@@ -32,14 +32,19 @@
 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 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;
 
 /**
  * <p>Watch a path for changes.</p>
@@ -50,17 +55,19 @@
  */
 public class Watcher {
     private final Logger logger = LogManager.getLogger();
-    private final WatchScope scope;
     private final Path path;
+    private final WatchScope scope;
+    private volatile Approximation approximateOnOverflow = Approximation.ALL;
     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 static final Predicate<WatchEvent> TRUE_FILTER = e -> true;
+    private volatile Predicate<WatchEvent> eventFilter = TRUE_FILTER;
 
-    private Watcher(WatchScope scope, Path path) {
-        this.scope = scope;
+    private Watcher(Path path, WatchScope scope) {
         this.path = path;
+        this.scope = scope;
     }
 
     /**
@@ -87,9 +94,8 @@ public static Watcher watch(Path path, WatchScope scope) {
                 break;
             default:
                 throw new IllegalArgumentException("Unsupported scope: " + scope);
-
         }
-        return new Watcher(scope, path);
+        return new Watcher(path, scope);
     }
 
     /**
@@ -103,7 +109,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 +120,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);
@@ -135,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()}.
@@ -146,6 +168,22 @@ public Watcher withExecutor(Executor callbackHandler) {
         return this;
     }
 
+    /**
+     * Optionally configure which regular files/directories in the scope of the
+     * watch an <i>approximation</i> of synthetic events (of kinds
+     * {@link WatchEvent.Kind#CREATED}, {@link WatchEvent.Kind#MODIFIED}, and/or
+     * {@link WatchEvent.Kind#DELETED}) should be issued when an overflow event
+     * happens. If not defined before this watcher is started, the
+     * {@link Approximation#ALL} approach will be used.
+     * @param whichFiles Constant to indicate for which regular
+     * files/directories to approximate
+     * @return This watcher for optional method chaining
+     */
+    public Watcher onOverflow(Approximation whichFiles) {
+        this.approximateOnOverflow = whichFiles;
+        return this;
+    }
+
     /**
      * Start watch the path for events.
      * @return a subscription for the watch, when closed, new events will stop being registered to the worker pool.
@@ -157,33 +195,48 @@ public ActiveWatch start() throws IOException {
             throw new IllegalStateException("There is no onEvent handler defined");
         }
 
+        var h = applyApproximateOnOverflow();
+
         switch (scope) {
             case PATH_AND_CHILDREN: {
-                var result = new JDKDirectoryWatch(path, executor, eventHandler, false);
+                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, eventHandler);
+                var result = new JDKFileWatch(path, executor, h, eventFilter);
                 result.open();
                 return result;
             }
             default:
                 throw new IllegalStateException("Not supported yet");
         }
     }
+
+    private BiConsumer<EventHandlingWatch, WatchEvent> applyApproximateOnOverflow() {
+        switch (approximateOnOverflow) {
+            case NONE:
+                return eventHandler;
+            case ALL:
+                return eventHandler.andThen(new MemorylessRescanner(executor));
+            case DIFF:
+                return eventHandler.andThen(new IndexingRescanner(executor, path, scope));
+            default:
+                throw new UnsupportedOperationException("No event handler has been defined yet for this overflow policy");
+        }
+    }
 }