Initial pure JDK implementation

README.md

@@ -1,8 +1,59 @@
 # java-watch
-a java file watcher that works across platforms and supports recursion and single file watches
+a java file watcher that works across platforms and supports recursion, single file watches, and tries to make sure no events are missed.
 
+## Features
+
+Currently working features in java-watch:
+
+- Recursive watches, even if platform doesn't support it natively.
+- Recursive watches also work inside directories created after the watch started
+- On overflow events no **new** directories (and it's recursive files) are missed, modification events will however not be simulated
+- Single file watches
+- Multiple watches for the same directory are merged to avoid overloading the kernel
+- Events are process on a worker pool, which you can customize.
+
+Future features:
+
+- Avoid poll based watcher in macOS/OSX that only detects changes every 2 seconds
+- Support file watches natively in linux
+- Monitor only specific events (such as only CREATES)
+
+## Usage
+
+Import dependency in pom.xml:
+
+```xml
+<dependency>
+  <groupId>engineering.swat</groupId>
+  <artifactId>java-watch</artifactId>
+  <version>${java-watch-version}</version>
+</dependency>
+```
+
+Start using java-watch:
+
+```java
+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
+  .onEvent(watchEvent -> {
+    System.err.println(watchEvent);
+  });
+
+try(var active = watcherSetup.start()) {
+  System.out.println("Monitoring files, press any key to stop");
+  System.in.read();
+}
+// after active.close(), the watch is stopped and
+// no new events will be scheduled on the threadpool
+```
 
 ## Related work
+
+Before starting this library, we wanted to use existing libraries, but they all lacked proper support for recursive file watches or lacked configurability. This library now has a growing collection of tests and a small API that should allow for future improvements without breaking compatibility.
+
+The following section describes the related work research on the libraries and underlying limitations.
+
 After reading the documentation of the following discussion on file system watches:
 
 - [Paul Millr's nodejs chokidar](https://github.com/paulmillr/chokidar)

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

@@ -1,7 +1,22 @@
 package engineering.swat.watch;
 
+/**
+ * Configure the depth of the events you want to receive for a given path
+ */
 public enum WatchScope {
-    SINGLE,
-    INCLUDING_CHILDREN,
-    INCLUDING_ALL_DESCENDANTS
+    /**
+     *  Watch changes to a single file or (metadata of) a directory
+     */
+    PATH_ONLY,
+    /**
+     * Watch changes to (metadata of) a directory and its content,
+     * non-recursively. That is, changes to the content of nested directories
+     * are not watched.
+     */
+    PATH_AND_CHILDREN,
+    /**
+     * Watch changes to (metadata of) a directory and its content, recursively.
+     * That is, changes to the content of nested directories are also watched.
+     */
+    PATH_AND_ALL_DESCENDANTS
 }

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

@@ -35,7 +35,6 @@ public class Watcher {
     private Watcher(WatchScope scope, Path path) {
         this.scope = scope;
         this.path = path;
-        logger.info("Constructor logger for: {} at {} level", path, scope);
     }
 
     /**
@@ -49,13 +48,13 @@ public static Watcher watch(Path path, WatchScope scope) {
             throw new IllegalArgumentException("We can only watch absolute paths");
         }
         switch (scope) {
-            case INCLUDING_CHILDREN: // intended fallthrough
-            case INCLUDING_ALL_DESCENDANTS:
+            case PATH_AND_CHILDREN: // intended fallthrough
+            case PATH_AND_ALL_DESCENDANTS:
                 if (!Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS)) {
-                    throw new IllegalArgumentException("Only directories are supported for this scope");
+                    throw new IllegalArgumentException("Only directories are supported for this scope: " + scope);
                 }
                 break;
-            case SINGLE:
+            case PATH_ONLY:
                 if (Files.isSymbolicLink(path)) {
                     throw new IllegalArgumentException("Symlinks are not supported");
                 }
@@ -101,12 +100,12 @@ public Closeable start() throws IOException {
             throw new IllegalStateException("There is no onEvent handler defined");
         }
         switch (scope) {
-            case INCLUDING_CHILDREN: {
+            case PATH_AND_CHILDREN: {
                 var result = new JDKDirectoryWatcher(path, executor, this.eventHandler, false);
                 result.start();
                 return result;
             }
-            case INCLUDING_ALL_DESCENDANTS: {
+            case PATH_AND_ALL_DESCENDANTS: {
                 try {
                     var result = new JDKDirectoryWatcher(path, executor, this.eventHandler, true);
                     result.start();
@@ -120,7 +119,7 @@ public Closeable start() throws IOException {
                     return result;
                 }
             }
-            case SINGLE: {
+            case PATH_ONLY: {
                 var result = new JDKFileWatcher(path, executor, this.eventHandler);
                 result.start();
                 return result;

src/main/java/engineering/swat/watch/impl/BundledSubscription.java

@@ -3,11 +3,15 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
@@ -16,6 +20,7 @@
  * This is used (for example) to avoid multiple JDKPoller registries for the same path
  */
 public class BundledSubscription<Key extends @NonNull Object, Event extends @NonNull Object> implements ISubscribable<Key,Event> {
+    private static final Logger logger = LogManager.getLogger();
     private final ISubscribable<Key, Event> wrapped;
     private final ConcurrentMap<Key, Subscription<Event>> subscriptions = new ConcurrentHashMap<>();
 
@@ -27,18 +32,15 @@ public BundledSubscription(ISubscribable<Key, Event> wrapped) {
     private static class Subscription<R> implements Consumer<R> {
         private final List<Consumer<R>> consumers = new CopyOnWriteArrayList<>();
         private volatile @MonotonicNonNull Closeable toBeClosed;
+        private volatile boolean closed = false;
         Subscription() {
         }
 
-        public void setToBeClosed(Closeable closer) {
-            this.toBeClosed = closer;
-        }
-
-        public void add(Consumer<R> newConsumer) {
+        void add(Consumer<R> newConsumer) {
             consumers.add(newConsumer);
         }
 
-        public void remove(Consumer<R> existingConsumer) {
+        void remove(Consumer<R> existingConsumer) {
             consumers.remove(existingConsumer);
         }
 
@@ -52,51 +54,58 @@ public void accept(R t) {
         boolean hasActiveConsumers() {
             return !consumers.isEmpty();
         }
-
-
     }
 
     @Override
     public Closeable subscribe(Key target, Consumer<Event> eventListener) throws IOException {
-        var active = this.subscriptions.computeIfAbsent(target, t -> new Subscription<>());
-        boolean first = false;
-        if (active.toBeClosed == null) {
-            // we just added a new one
-            // so lets take a lock on it, and try to be the one that gets to initialize it
+        while (true) {
+            Subscription<Event> active = this.subscriptions.computeIfAbsent(target, t -> new Subscription<>());
+            // after this, there will only be 1 instance of active subscription in the map.
+            // but we might have a race with remove, which can close the subscript between our get and our addition
+            // since this code is very hard to get right without locks, and shouldn't be run too often
+            // we take a big lock around the subscription management
             synchronized(active) {
-                // now lock on it to make sure nobo
-                if (active.toBeClosed == null) {
-                    first = true;
-                    active.add(eventListener); // we know we already have the lock, and we need to do this before we register the watch
-                    var newSubscriptions = wrapped.subscribe(target, active);
-                    active.setToBeClosed(newSubscriptions);
+                if (active.closed) {
+                    // we lost the race with closing the subscription, so we retry
+                    continue;
                 }
-                else {
+                active.add(eventListener);
+                if (active.toBeClosed == null) {
+                    // the watch is not active yet, and we were the first to get the lock
+                    active.toBeClosed = wrapped.subscribe(target, active);
                 }
             }
-        }
-        // at this point we have to be sure that we're not the first to in the list
-        // since we might have won the race on the compute, but lost the race
-        if (!first) {
-            active.add(eventListener);
-        }
-        return () -> {
-            active.remove(eventListener);
-            if (!active.hasActiveConsumers()) {
-                subscriptions.remove(target);
-                if (active.hasActiveConsumers()) {
-                    // we lost the race, someone else added something again
-                    // so we put it back in the list
-                    subscriptions.put(target, active);
+            return () -> {
+                boolean scheduleClose = false;
+                synchronized(active) {
+                    active.remove(eventListener);
+                    scheduleClose = !active.hasActiveConsumers() && !active.closed;
                 }
-                else {
-                    if (active.toBeClosed != null) {
-                        active.toBeClosed.close();
-                    }
+                if (scheduleClose) {
+                    // to avoid hammering the system with closes & registers in a short periode
+                    // we schedule the cleanup of watches in the background, when even after a small delay
+                    // nobody is interested in a certain file anymore
+                    CompletableFuture
+                        .delayedExecutor(100, TimeUnit.MILLISECONDS)
+                        .execute(() -> {
+                            synchronized(active) {
+                                if (!active.hasActiveConsumers() && !active.closed) {
+                                    // still ready to be closed
+                                    active.closed = true;
+                                    this.subscriptions.remove(target, active);
+                                    if (active.toBeClosed != null) {
+                                        try {
+                                            active.toBeClosed.close();
+                                        } catch (IOException e) {
+                                            logger.error("Unhandled exception while closing the watcher for {} in the background", target, e);
+                                        }
+                                    }
+                                }
+                            }
+                        });
                 }
-            }
-        };
-
+            };
+        }
     }
 
 

src/main/java/engineering/swat/watch/impl/JDKPoller.java

@@ -3,6 +3,7 @@
 import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
 import static java.nio.file.StandardWatchEventKinds.ENTRY_DELETE;
 import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
+import static java.nio.file.StandardWatchEventKinds.OVERFLOW;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -92,7 +93,7 @@ public static Closeable register(SubscriptionKey path, Consumer<List<WatchEvent<
         try {
             return CompletableFuture.supplyAsync(() -> {
                 try {
-                    WatchEvent.Kind<?>[] kinds = new WatchEvent.Kind[]{ ENTRY_CREATE, ENTRY_MODIFY, ENTRY_MODIFY, ENTRY_DELETE };
+                    WatchEvent.Kind<?>[] kinds = new WatchEvent.Kind[]{ ENTRY_CREATE, ENTRY_MODIFY, OVERFLOW, ENTRY_DELETE };
                     if (path.isRecursive()) {
                         return path.getPath().register(service, kinds, ExtendedWatchEventModifier.FILE_TREE);
                     }
@@ -109,8 +110,9 @@ public static Closeable register(SubscriptionKey path, Consumer<List<WatchEvent<
                     @Override
                     public void close() throws IOException {
                         logger.debug("Closing watch for: {}", path);
-                        key.cancel();
-                        watchers.remove(key);
+                        if (watchers.remove(key, changesHandler)) {
+                            key.cancel();
+                        }
                     }
                 };
             })

src/main/java/engineering/swat/watch/impl/SubscriptionKey.java

@@ -36,4 +36,9 @@ public boolean equals(@Nullable Object obj) {
     public int hashCode() {
         return Objects.hash(path, recursive);
     }
+
+    @Override
+    public String toString() {
+        return path.toString() + (recursive ? "[recursive]" : "");
+    }
 }

src/test/java/engineering/swat/watch/DeleteLockTests.java

@@ -61,7 +61,7 @@ private void deleteAndVerify(Path target, WatchScope scope) throws IOException {
     void watchedFileCanBeDeleted() throws IOException {
         deleteAndVerify(
             testDir.getTestFiles().get(0),
-            WatchScope.SINGLE
+            WatchScope.PATH_ONLY
         );
     }
 
@@ -70,7 +70,7 @@ void watchedFileCanBeDeleted() throws IOException {
     void watchedDirectoryCanBeDeleted() throws IOException {
         deleteAndVerify(
             testDir.getTestDirectory(),
-            WatchScope.INCLUDING_CHILDREN
+            WatchScope.PATH_AND_CHILDREN
         );
     }
 
@@ -79,7 +79,7 @@ void watchedDirectoryCanBeDeleted() throws IOException {
     void watchedRecursiveDirectoryCanBeDeleted() throws IOException {
         deleteAndVerify(
             testDir.getTestDirectory(),
-            WatchScope.INCLUDING_ALL_DESCENDANTS
+            WatchScope.PATH_AND_ALL_DESCENDANTS
         );
     }
 }

src/test/java/engineering/swat/watch/RecursiveWatchTests.java

@@ -47,7 +47,7 @@ void newDirectoryWithFilesChangesDetected() throws IOException {
         var target = new AtomicReference<Path>();
         var created = new AtomicBoolean(false);
         var changed = new AtomicBoolean(false);
-        var watchConfig = Watcher.watch(testDir.getTestDirectory(), WatchScope.INCLUDING_ALL_DESCENDANTS)
+        var watchConfig = Watcher.watch(testDir.getTestDirectory(), WatchScope.PATH_AND_ALL_DESCENDANTS)
             .onEvent(ev -> {
                     logger.debug("Event received: {}", ev);
                     if (ev.calculateFullPath().equals(target.get())) {
@@ -79,7 +79,7 @@ void newDirectoryWithFilesChangesDetected() throws IOException {
     void correctRelativePathIsReported() throws IOException {
         Path relative = Path.of("a","b", "c", "d.txt");
         var seen = new AtomicBoolean(false);
-        var watcher = Watcher.watch(testDir.getTestDirectory(), WatchScope.INCLUDING_ALL_DESCENDANTS)
+        var watcher = Watcher.watch(testDir.getTestDirectory(), WatchScope.PATH_AND_ALL_DESCENDANTS)
             .onEvent(ev -> {
                 logger.debug("Seen event: {}", ev);
                 if (ev.getRelativePath().equals(relative)) {
@@ -104,7 +104,7 @@ void deleteOfFileInDirectoryShouldBeVisible() throws IOException, InterruptedExc
             .findAny()
             .orElseThrow();
         var seen = new AtomicBoolean(false);
-        var watchConfig = Watcher.watch(target.getParent(), WatchScope.INCLUDING_CHILDREN)
+        var watchConfig = Watcher.watch(target.getParent(), WatchScope.PATH_AND_CHILDREN)
             .onEvent(ev -> {
                 if (ev.getKind() == Kind.DELETED && ev.calculateFullPath().equals(target)) {
                     seen.set(true);

src/test/java/engineering/swat/watch/SingleDirectoryTests.java

@@ -40,7 +40,7 @@ static void setupEverything() {
     void deleteOfFileInDirectoryShouldBeVisible() throws IOException, InterruptedException {
         var target = testDir.getTestFiles().get(0);
         var seen = new AtomicBoolean(false);
-        var watchConfig = Watcher.watch(target.getParent(), WatchScope.INCLUDING_CHILDREN)
+        var watchConfig = Watcher.watch(target.getParent(), WatchScope.PATH_AND_CHILDREN)
             .onEvent(ev -> {
                 if (ev.getKind() == Kind.DELETED && ev.calculateFullPath().equals(target)) {
                     seen.set(true);

src/test/java/engineering/swat/watch/SingleFileTests.java

@@ -39,7 +39,7 @@ void singleFileShouldNotTriggerOnOtherFilesInSameDir() throws IOException, Inter
         var target = testDir.getTestFiles().get(0);
         var seen = new AtomicBoolean(false);
         var others = new AtomicBoolean(false);
-        var watchConfig = Watcher.watch(target, WatchScope.SINGLE)
+        var watchConfig = Watcher.watch(target, WatchScope.PATH_ONLY)
             .onEvent(ev -> {
                 if (ev.calculateFullPath().equals(target)) {
                     seen.set(true);
@@ -68,7 +68,7 @@ void singleFileThatMonitorsOnlyADirectory() throws IOException, InterruptedExcep
         var target = testDir.getTestDirectory();
         var seen = new AtomicBoolean(false);
         var others = new AtomicBoolean(false);
-        var watchConfig = Watcher.watch(target, WatchScope.SINGLE)
+        var watchConfig = Watcher.watch(target, WatchScope.PATH_ONLY)
             .onEvent(ev -> {
                 if (ev.calculateFullPath().equals(target)) {
                     seen.set(true);