Initial pure JDK implementation

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

@@ -32,7 +32,7 @@ public enum Kind {
         DELETED,
         /**
          * Rare event where there were so many file events, that the kernel lost a few.
-         * In that case you'll have to consider the whole directory (and it's sub directories) as modified.
+         * In that case you'll have to consider the whole directory (and its sub directories) as modified.
          * The library will try and send events for new and deleted files, but it won't be able to detect modified files.
          */
         OVERFLOW

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

@@ -0,0 +1,7 @@
+package engineering.swat.watch;
+
+public enum WatchScope {
+    SINGLE,
+    INCLUDING_CHILDREN,
+    INCLUDING_ALL_DESCENDANTS
+}

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

@@ -24,71 +24,47 @@
  */
 public class Watcher {
     private final Logger logger = LogManager.getLogger();
-    private final WatcherKind kind;
+    private final WatchScope scope;
     private final Path path;
-    private Executor executor = CompletableFuture::runAsync;
+    private volatile Executor executor = CompletableFuture::runAsync;
 
     private static final Consumer<WatchEvent> NULL_HANDLER = p -> {};
-    private Consumer<WatchEvent> eventHandler = NULL_HANDLER;
+    private volatile Consumer<WatchEvent> eventHandler = NULL_HANDLER;
 
 
-    private Watcher(WatcherKind kind, Path path) {
-        this.kind = kind;
+    private Watcher(WatchScope scope, Path path) {
+        this.scope = scope;
         this.path = path;
-        logger.info("Constructor logger for: {} at {} level", path, kind);
-    }
-
-    private enum WatcherKind {
-        FILE,
-        DIRECTORY,
-        RECURSIVE_DIRECTORY
+        logger.info("Constructor logger for: {} at {} level", path, scope);
     }
 
     /**
-     * Request a watcher for a single path (file or directory).
-     * If it's a file, depending on the platform this will watch the whole directory and filter the results, or only watch a single file.
-     * @param path a single path entry, either a file or a directory
-     * @return a watcher that only fires events related to the requested path
-     * @throws IOException in case the path is not absolute
+     * Watch a path for updates, optionally also get events for its children/descendants
+     * @param path which absolute path to monitor, can be a file or a directory, but has to be absolute
+     * @param scope for directories you can also choose to monitor it's direct children or all it's descendants
+     * @throws IllegalArgumentException in case a path is not supported (in relation to the scope)
      */
-    public static Watcher single(Path path) throws IOException {
+    public static Watcher watch(Path path, WatchScope scope) {
         if (!path.isAbsolute()) {
-            throw new IOException("We can only watch absolute paths");
-        }
-        return new Watcher(WatcherKind.FILE, path);
-    }
-
-    /**
-     * Request a watcher for a directory, getting events for its direct children.
-     * @param path a directory to monitor for changes
-     * @return a watcher that fires events for any of the direct children (and its self).
-     * @throws IOException in cas the path is not absolute or it's not an directory
-     */
-    public static Watcher singleDirectory(Path path) throws IOException {
-        if (!path.isAbsolute()) {
-            throw new IOException("We can only watch absolute paths");
-        }
-        if (!Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS)) {
-            throw new IOException("Only directories are supported");
+            throw new IllegalArgumentException("We can only watch absolute paths");
         }
-        return new Watcher(WatcherKind.DIRECTORY, path);
-    }
+        switch (scope) {
+            case INCLUDING_CHILDREN: // intended fallthrough
+            case INCLUDING_ALL_DESCENDANTS:
+                if (!Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS)) {
+                    throw new IllegalArgumentException("Only directories are supported for this scope");
+                }
+                break;
+            case SINGLE:
+                if (Files.isSymbolicLink(path)) {
+                    throw new IllegalArgumentException("Symlinks are not supported");
+                }
+                break;
+            default:
+                throw new IllegalArgumentException("Unsupported scope: " + scope);
 
-    /**
-     * Request a watcher for a directory, getting events for all of its children. Even those added after the watch started.
-     * On some platforms, this can be quite expansive, so be sure you want this.
-     * @param path a directory to monitor for changes
-     * @return a watcher that fires events for any of its children (and its self).
-     * @throws IOException in case the path is not absolute or it's not an directory
-     */
-    public static Watcher recursiveDirectory(Path path) throws IOException {
-        if (!path.isAbsolute()) {
-            throw new IOException("We can only watch absolute paths");
-        }
-        if (!Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS)) {
-            throw new IOException("Only directories are supported");
         }
-        return new Watcher(WatcherKind.RECURSIVE_DIRECTORY, path);
+        return new Watcher(scope, path);
     }
 
     /**
@@ -118,24 +94,33 @@ public Watcher withExecutor(Executor callbackHandler) {
      * Start watch the path for events.
      * @return a subscription for the watch, when closed, new events will stop being registered to the worker pool.
      * @throws IOException in case the starting of the watcher caused an underlying IO exception
-     * @throws IllegalStateException the watchers is not configured correctly (for example, missing {@link #onEvent(Consumer)})
+     * @throws IllegalStateException the watchers is not configured correctly (for example, missing {@link #onEvent(Consumer)}, or a watcher is started twice)
      */
-    public Closeable start() throws IOException, IllegalStateException {
+    public Closeable start() throws IOException {
         if (this.eventHandler == NULL_HANDLER) {
             throw new IllegalStateException("There is no onEvent handler defined");
         }
-        switch (kind) {
-            case DIRECTORY: {
-                var result = new JDKDirectoryWatcher(path, executor, this.eventHandler);
+        switch (scope) {
+            case INCLUDING_CHILDREN: {
+                var result = new JDKDirectoryWatcher(path, executor, this.eventHandler, false);
                 result.start();
                 return result;
             }
-            case RECURSIVE_DIRECTORY: {
-                var result = new JDKRecursiveDirectoryWatcher(path, executor, this.eventHandler);
-                result.start();
-                return result;
+            case INCLUDING_ALL_DESCENDANTS: {
+                try {
+                    var result = new JDKDirectoryWatcher(path, executor, this.eventHandler, true);
+                    result.start();
+                    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 JDKRecursiveDirectoryWatcher(path, executor, this.eventHandler);
+                    result.start();
+                    return result;
+                }
             }
-            case FILE: {
+            case SINGLE: {
                 var result = new JDKFileWatcher(path, executor, this.eventHandler);
                 result.start();
                 return result;

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

@@ -11,33 +11,35 @@
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
-public class BundledSubscription<A extends @NonNull Object, R extends @NonNull Object> implements ISubscribable<A,R> {
-    private final ISubscribable<A, R> around;
-    private final ConcurrentMap<A, Subscription<R>> subscriptions = new ConcurrentHashMap<>();
+/**
+ * This is an internal class where we can join multiple subscriptions to the same target by only taking 1 actual subscription but forwarding them to all the interested parties.
+ * 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 final ISubscribable<Key, Event> wrapped;
+    private final ConcurrentMap<Key, Subscription<Event>> subscriptions = new ConcurrentHashMap<>();
 
-    public BundledSubscription(ISubscribable<A, R> around) {
-        this.around = around;
+    public BundledSubscription(ISubscribable<Key, Event> wrapped) {
+        this.wrapped = wrapped;
 
     }
 
     private static class Subscription<R> implements Consumer<R> {
         private final List<Consumer<R>> consumers = new CopyOnWriteArrayList<>();
-        private volatile @MonotonicNonNull Closeable closer;
-        Subscription(Consumer<R> initialConsumer) {
-            consumers.add(initialConsumer);
+        private volatile @MonotonicNonNull Closeable toBeClosed;
+        Subscription() {
         }
 
-        public void setCloser(Closeable closer) {
-            this.closer = closer;
+        public void setToBeClosed(Closeable closer) {
+            this.toBeClosed = closer;
         }
 
-        void add(Consumer<R> newConsumer) {
+        public void add(Consumer<R> newConsumer) {
             consumers.add(newConsumer);
         }
 
-        synchronized boolean remove(Consumer<R> existingConsumer) {
+        public void remove(Consumer<R> existingConsumer) {
             consumers.remove(existingConsumer);
-            return consumers.isEmpty();
         }
 
         @Override
@@ -55,38 +57,41 @@ boolean hasActiveConsumers() {
     }
 
     @Override
-    public Closeable subscribe(A target, Consumer<R> eventListener) throws IOException {
-        var active = this.subscriptions.get(target);
-        if (active == null) {
-            active = new Subscription<>(eventListener);
-            var newSubscriptions = around.subscribe(target, active);
-            active.setCloser(newSubscriptions);
-            var lostRace = this.subscriptions.putIfAbsent(target, active);
-            if (lostRace != null) {
-                try {
-                    newSubscriptions.close();
-                } catch (IOException _ignore) {
-                    // ignore
+    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
+            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);
+                }
+                else {
                 }
-                lostRace.add(eventListener);
-                active = lostRace;
             }
         }
-        else {
+        // 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);
         }
-        var finalActive = active;
         return () -> {
-            if (finalActive.remove(eventListener)) {
+            active.remove(eventListener);
+            if (!active.hasActiveConsumers()) {
                 subscriptions.remove(target);
-                if (finalActive.hasActiveConsumers()) {
+                if (active.hasActiveConsumers()) {
                     // we lost the race, someone else added something again
                     // so we put it back in the list
-                    subscriptions.put(target, finalActive);
+                    subscriptions.put(target, active);
                 }
                 else {
-                    if (finalActive.closer != null) {
-                        finalActive.closer.close();
+                    if (active.toBeClosed != null) {
+                        active.toBeClosed.close();
                     }
                 }
             }

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

@@ -5,6 +5,6 @@
 import java.util.function.Consumer;
 
 @FunctionalInterface
-public interface ISubscribable<A, R> {
-    Closeable subscribe(A target, Consumer<R> eventListener) throws IOException;
+public interface ISubscribable<Key, Event> {
+    Closeable subscribe(Key target, Consumer<Event> eventListener) throws IOException;
 }

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

@@ -21,28 +21,35 @@ public class JDKDirectoryWatcher implements Closeable {
     private final Executor exec;
     private final Consumer<WatchEvent> eventHandler;
     private volatile @MonotonicNonNull Closeable activeWatch;
+    private final boolean nativeRecursive;
 
-    private static final BundledSubscription<Path, List<java.nio.file.WatchEvent<?>>>
+    private static final BundledSubscription<SubscriptionKey, List<java.nio.file.WatchEvent<?>>>
         BUNDLED_JDK_WATCHERS = new BundledSubscription<>(JDKPoller::register);
 
     public JDKDirectoryWatcher(Path directory, Executor exec, Consumer<WatchEvent> eventHandler) {
+        this(directory, exec, eventHandler, false);
+    }
+
+    public JDKDirectoryWatcher(Path directory, Executor exec, Consumer<WatchEvent> eventHandler, boolean nativeRecursive) {
         this.directory = directory;
         this.exec = exec;
         this.eventHandler = eventHandler;
+        this.nativeRecursive = nativeRecursive;
     }
 
+
     synchronized boolean safeStart() throws IOException {
         if (activeWatch != null) {
             return false;
         }
-        activeWatch = BUNDLED_JDK_WATCHERS.subscribe(directory, this::handleChanges);
+        activeWatch = BUNDLED_JDK_WATCHERS.subscribe(new SubscriptionKey(directory, nativeRecursive), this::handleChanges);
         return true;
     }
 
     public void start() throws IOException {
         try {
             if (!safeStart()) {
-                throw new IOException("Cannot start a watcher twice");
+                throw new IllegalStateException("Cannot start a watcher twice");
             }
             logger.debug("Started watch for: {}", directory);
         } catch (IOException e) {
@@ -86,7 +93,7 @@ else if (ev.kind() == StandardWatchEventKinds.OVERFLOW) {
     }
 
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
         if (activeWatch != null) {
             logger.debug("Closing watch for: {}", this.directory);
             activeWatch.close();

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

@@ -14,6 +14,8 @@
 
 /**
  * It's not possible to monitor a single file (or directory), so we have to find a directory watcher, and connect to that
+ *
+ * Note that you should take care to call start only once.
  */
 public class JDKFileWatcher implements Closeable {
     private final Logger logger = LogManager.getLogger();
@@ -34,23 +36,27 @@ public JDKFileWatcher(Path file, Executor exec, Consumer<WatchEvent> eventHandle
         this.eventHandler = eventHandler;
     }
 
+    /**
+     * Start the file watcher, but only do it once
+     * @throws IOException
+     */
     public void start() throws IOException {
         try {
+            var dir = file.getParent();
+            if (dir == null) {
+                throw new IllegalArgumentException("cannot watch a single entry that is on the root");
+
+            }
+            assert !dir.equals(file);
+            JDKDirectoryWatcher parentWatch;
             synchronized(this) {
                 if (activeWatch != null) {
                     throw new IOException("Cannot start an already started watch");
                 }
-                var dir = file.getParent();
-                if (dir == null) {
-                    throw new IllegalArgumentException("cannot watch a single entry that is on the root");
-
-                }
-                assert !dir.equals(file);
-                var parentWatch = new JDKDirectoryWatcher(dir, exec, this::filter);
-                activeWatch = parentWatch;
+                activeWatch = parentWatch = new JDKDirectoryWatcher(dir, exec, this::filter);
                 parentWatch.start();
-                logger.debug("Started file watch for {} (in reality a watch on {}): {}", file, dir, parentWatch);
             }
+            logger.debug("Started file watch for {} (in reality a watch on {}): {}", file, dir, parentWatch);
 
         } catch (IOException e) {
             throw new IOException("Could not register file watcher for: " + file, e);
@@ -59,12 +65,12 @@ public void start() throws IOException {
 
     private void filter(WatchEvent event) {
         if (fileName.equals(event.getRelativePath())) {
-            exec.execute(() -> eventHandler.accept(event));
+            eventHandler.accept(event);
         }
     }
 
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
         if (activeWatch != null) {
             activeWatch.close();
         }