#36 added FileObserver

assira.core/src/main/java/net/ranides/assira/io/FileObserver.java

@@ -0,0 +1,212 @@
+package net.ranides.assira.io;
+
+import lombok.extern.slf4j.Slf4j;
+
+import net.ranides.assira.collection.query.CQuery;
+import net.ranides.assira.events.EventDispatcher;
+import net.ranides.assira.events.EventListener;
+import net.ranides.assira.events.EventRouter;
+import net.ranides.assira.trace.ExceptionUtils;
+
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.WatchEvent;
+import java.nio.file.WatchKey;
+import java.nio.file.WatchService;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+import static java.nio.file.StandardWatchEventKinds.*;
+
+@Slf4j
+public class FileObserver implements AutoCloseable {
+
+    private final WatchService service;
+
+    private final Map<WatchKey, Path> observers = new ConcurrentHashMap<>();
+
+    private final Map<Path, WatchKey> directories = new ConcurrentHashMap<>();
+
+    private final EventRouter router;
+
+    public FileObserver() throws IOException {
+        this(new EventDispatcher());
+    }
+
+    public FileObserver(EventRouter router) throws IOException {
+        this.service = FileSystems.getDefault().newWatchService();
+        this.router = router;
+    }
+
+    @Override
+    public void close() {
+        for (WatchKey w : observers.keySet()) {
+            w.cancel();
+        }
+        observers.clear();
+        directories.clear();
+    }
+
+    public FileObserver addEventListener(EventListener<? super FileObserverEvent> listener) {
+        router.addEventListener(FileObserverEvent.class, listener);
+        return this;
+    }
+
+    public <T extends FileObserverEvent> FileObserver addEventListener(Class<T> event, EventListener<? super T> listener) {
+        router.addEventListener(event, listener);
+        return this;
+    }
+
+    public FileObserver onCreate(Consumer<Path> create) {
+        addEventListener(FileObserverEvent.Create.class, event -> create.accept(event.path()));
+        return this;
+    }
+
+    public FileObserver onDelete(Consumer<Path> delete) {
+        addEventListener(FileObserverEvent.Delete.class, event -> delete.accept(event.path()));
+        return this;
+    }
+
+    public FileObserver onModify(Consumer<Path> modify) {
+        addEventListener(FileObserverEvent.Modify.class, event -> modify.accept(event.path()));
+        return this;
+    }
+
+    public FileObserver observe(Path directory) {
+        try {
+            Path path = PathUtils.normalize(directory);
+            log.debug("Observe directory {}", path);
+            Files.list(path).forEach(p -> {
+                if(Files.isDirectory(p)) {
+                    observe(p);
+                }
+                else {
+                    try {
+                        router.signalEvent(FileObserverEvent.create(p));
+                    } catch (Exception cause) {
+                        log.error(cause.getMessage(), cause);
+                    }
+                }
+            });
+            WatchKey key = path.register(service, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
+            observers.put(key, path);
+            directories.put(path, key);
+        } catch (IOException e) {
+            throw ExceptionUtils.rethrow(e);
+        }
+        return this;
+    }
+
+    public void unobserve(Path path) {
+        Path normalized = PathUtils.normalize(path);
+        CQuery.from(directories.keySet())
+            .filter(k -> k.startsWith(normalized))
+            .fetch()
+            .forEach(this::cancel);
+    }
+
+    public boolean process() {
+        boolean success = true;
+
+        WatchKey key;
+        while(null != (key= service.poll())) {
+            log.trace("Change inside directory: {}", observers.get(key));
+            success &= process(key);
+        }
+
+        observers.entrySet().removeIf(o -> !o.getKey().isValid());
+
+        return success;
+    }
+
+    private boolean process(WatchKey key) {
+        Map<Path,  WatchEvent.Kind<?>> paths = new HashMap<>();
+
+        for (WatchEvent<?> event : key.pollEvents()) {
+            WatchEvent.Kind<?> kind = event.kind();
+            if (kind == OVERFLOW) {
+                continue;
+            }
+
+            Path dirinfo = observers.get(key);
+            Path filename = (Path) event.context();
+            Path filepath = dirinfo.resolve(filename);
+
+            log.trace("File event: {} {}", kind, filepath);
+
+            if(kind == ENTRY_DELETE && directories.containsKey(filepath)) {
+                cancel(filepath);
+                continue;
+            }
+
+            if (Files.isDirectory(filepath)) {
+                if(kind == ENTRY_CREATE) {
+                    observe(filepath);
+                } else {
+                    // change events can be ignored
+                    // delete events won't be here, because Files.isDirectory returns false for non-existent paths
+                    continue;
+                }
+            }
+
+            paths.put(filepath, merge(paths.get(filepath), kind));
+        }
+
+        for(Map.Entry<Path, WatchEvent.Kind<?>> entry : paths.entrySet()) {
+            Path path = entry.getKey();
+            WatchEvent.Kind<?> kind = entry.getValue();
+
+            log.trace("File event process: {} {}", kind, path);
+
+            try {
+                if (kind == ENTRY_CREATE) {
+                    if(Files.exists(path)) {
+                        router.handleEvent(FileObserverEvent.create(path));
+                    } else {
+                        log.trace("ENTRY_CREATE for not-existent path: {}", path);
+                    }
+                } else if (kind == ENTRY_DELETE) {
+                    router.handleEvent(FileObserverEvent.delete(path));
+                } else if (kind == ENTRY_MODIFY) {
+                    if(Files.exists(path)) {
+                        router.handleEvent(FileObserverEvent.modify(path));
+                    } else {
+                        // BUG in JDK or Windows:
+                        // if directory is removed, then we will receive ENTRY_MODIFY for every file inside
+                        // it's weird, because we should receive ENTRY_DELETE
+                        log.trace("ENTRY_MODIFY for not-existent path: {}", path);
+                        router.handleEvent(FileObserverEvent.delete(path));
+                    }
+                } else {
+                    log.warn("Unknown file system event {} {}", kind.name(), path);
+                }
+            } catch (Exception cause) {
+                log.error(cause.getMessage(), cause);
+            }
+        }
+
+        return key.reset();
+    }
+
+    private void cancel(Path filepath) {
+        log.debug("Remove directory: " + filepath);
+        directories.remove(filepath).cancel();
+    }
+
+    private WatchEvent.Kind<?> merge(WatchEvent.Kind<?> prev, WatchEvent.Kind<?> next) {
+        //        NEW : C M D
+        // create     : - - -
+        // modify     : M - -
+        // delete     : M - -
+        if( next == ENTRY_CREATE && (prev == ENTRY_MODIFY || prev == ENTRY_DELETE)) {
+            return ENTRY_MODIFY;
+        } else {
+            return next;
+        }
+    }
+
+}

assira.core/src/main/java/net/ranides/assira/io/FileObserverEvent.java

@@ -0,0 +1,49 @@
+package net.ranides.assira.io;
+
+import net.ranides.assira.events.Event;
+
+import java.nio.file.Path;
+
+public abstract class FileObserverEvent implements Event {
+
+    private final Path path;
+
+    protected FileObserverEvent(Path path) {
+        this.path = path;
+    }
+
+    public Path path() {
+        return path;
+    }
+
+    public static Create create(Path path) {
+        return new Create(path);
+    }
+
+    public static Modify modify(Path path) {
+        return new Modify(path);
+    }
+
+    public static Delete delete(Path path) {
+        return new Delete(path);
+    }
+
+    public static class Create extends FileObserverEvent {
+        protected Create(Path path) {
+            super(path);
+        }
+    }
+
+    public static class Modify extends FileObserverEvent {
+        protected Modify(Path path) {
+            super(path);
+        }
+    }
+
+    public static class Delete extends FileObserverEvent {
+        protected Delete(Path path) {
+            super(path);
+        }
+    }
+
+}