From 976e29947c14573aef361b37a6e6a0f96dbfc0e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Prante?= Date: Mon, 9 Oct 2023 11:09:56 +0200 Subject: [PATCH] add file follow event test --- .../event/io/file/DefaultFileFollowEvent.java | 12 ++++ .../xbib/event/io/file/FileFollowEvent.java | 4 ++ .../event/io/file/FileFollowEventManager.java | 2 +- .../event/io/file/FileFollowEventService.java | 71 ++++++++----------- .../io/file/FileFollowEventManagerTest.java | 23 ++++-- .../io/file/TestFileFollowEventConsumer.java | 2 +- 6 files changed, 67 insertions(+), 47 deletions(-) diff --git a/src/main/java/org/xbib/event/io/file/DefaultFileFollowEvent.java b/src/main/java/org/xbib/event/io/file/DefaultFileFollowEvent.java index 8bd4cf6..3250895 100644 --- a/src/main/java/org/xbib/event/io/file/DefaultFileFollowEvent.java +++ b/src/main/java/org/xbib/event/io/file/DefaultFileFollowEvent.java @@ -8,6 +8,8 @@ public class DefaultFileFollowEvent extends DefaultEvent implements FileFollowEv private Path path; + private String content; + @Override public void setPath(Path path) { this.path = path; @@ -17,4 +19,14 @@ public class DefaultFileFollowEvent extends DefaultEvent implements FileFollowEv public Path getPath() { return path; } + + @Override + public void setContent(String content) { + this.content = content; + } + + @Override + public String getContent() { + return content; + } } diff --git a/src/main/java/org/xbib/event/io/file/FileFollowEvent.java b/src/main/java/org/xbib/event/io/file/FileFollowEvent.java index 6b13990..0abf8c8 100644 --- a/src/main/java/org/xbib/event/io/file/FileFollowEvent.java +++ b/src/main/java/org/xbib/event/io/file/FileFollowEvent.java @@ -9,4 +9,8 @@ public interface FileFollowEvent extends Event { void setPath(Path path); Path getPath(); + + void setContent(String content); + + String getContent(); } diff --git a/src/main/java/org/xbib/event/io/file/FileFollowEventManager.java b/src/main/java/org/xbib/event/io/file/FileFollowEventManager.java index b786956..273898f 100644 --- a/src/main/java/org/xbib/event/io/file/FileFollowEventManager.java +++ b/src/main/java/org/xbib/event/io/file/FileFollowEventManager.java @@ -25,7 +25,7 @@ public class FileFollowEventManager { ExecutorService executorService, ClassLoader classLoader) { this.eventServiceMap = new LinkedHashMap<>(); - for (Map.Entry followfiles : settings.getGroups("filefollow").entrySet()) { + for (Map.Entry followfiles : settings.getGroups("event.filefollow").entrySet()) { Settings definition = followfiles.getValue(); String baseStr = definition.get("base"); String patternStr = definition.get("pattern"); diff --git a/src/main/java/org/xbib/event/io/file/FileFollowEventService.java b/src/main/java/org/xbib/event/io/file/FileFollowEventService.java index 09460b0..79295c4 100644 --- a/src/main/java/org/xbib/event/io/file/FileFollowEventService.java +++ b/src/main/java/org/xbib/event/io/file/FileFollowEventService.java @@ -5,6 +5,7 @@ import org.xbib.settings.Settings; import java.io.Closeable; import java.io.IOException; +import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.channels.SeekableByteChannel; @@ -24,13 +25,12 @@ import java.util.logging.Level; import java.util.logging.Logger; import java.util.regex.Matcher; import java.util.regex.Pattern; +import java.util.stream.Stream; public class FileFollowEventService implements Callable, Closeable { private static final Logger logger = Logger.getLogger(FileFollowEventService.class.getName()); - private final Settings settings; - private final EventBus eventBus; private final Path base; @@ -41,12 +41,8 @@ public class FileFollowEventService implements Callable, Closeable { private final WatchService watchService; - private final WatchKey watchKey; - private final Map fileSizes; - private int eventCount; - private volatile boolean keepWatching; public FileFollowEventService(Settings settings, @@ -54,7 +50,6 @@ public class FileFollowEventService implements Callable, Closeable { Path base, Pattern pattern, Class eventClass) throws IOException { - this.settings = settings; this.eventBus = eventBus; this.base = base; this.pattern = pattern; @@ -63,9 +58,9 @@ public class FileFollowEventService implements Callable, Closeable { this.watchService = fileSystem.newWatchService(); WatchEvent.Kind[] kinds = new WatchEvent.Kind[1]; kinds[0] = StandardWatchEventKinds.ENTRY_MODIFY; - this.watchKey = base.register(watchService, kinds); - // limit file size memory to 32 files - this.fileSizes = new LimitedMap<>(32); + base.register(watchService, kinds); + this.fileSizes = new LinkedHashMap<>(); + fillFileSizes(base, pattern); this.keepWatching = true; } @@ -85,6 +80,7 @@ public class FileFollowEventService implements Callable, Closeable { Path path = pathWatchEvent.context(); Matcher matcher = pattern.matcher(path.toString()); if (!matcher.matches()) { + logger.log(Level.FINE, "no match of " + path + " to pattern " + pattern); continue; } Path p = base.resolve(path); @@ -92,22 +88,14 @@ public class FileFollowEventService implements Callable, Closeable { long lastSize = fileSizes.getOrDefault(p, 0L); long currentSize = p.toFile().length(); fileSizes.put(p, currentSize); - // We have no idea where to start reading if this is the first time. - // Avoid reading the whole file at first time, read only real diff. - // This means first event is swallowed! - if (lastSize > 0L) { - String content = readRange(channel, lastSize, currentSize); - // split content by line, this allows pattern matching without preprocessing in worker - for (String line : content.split("\n")) { - FileFollowEvent event = eventClass.getDeclaredConstructor().newInstance(); - event.setKey(path.toString()); - event.setPath(base); - event.setMap(new LinkedHashMap<>()); - event.getMap().putAll(settings.getAsStructuredMap()); - event.getMap().put("content", line); - eventBus.post(event); - eventCount++; - } + String content = readRange(channel, lastSize, currentSize); + // split content by line, this allows pattern matching without preprocessing in worker + for (String line : content.split("\n")) { + FileFollowEvent event = eventClass.getDeclaredConstructor().newInstance(); + event.setKey(base.toString()); + event.setPath(path); + event.setContent(line); + eventBus.post(event); } } } @@ -121,7 +109,7 @@ public class FileFollowEventService implements Callable, Closeable { } } } - return eventCount; + return 0; } @Override @@ -138,26 +126,29 @@ public class FileFollowEventService implements Callable, Closeable { fileChannel.position(from); int numRead = fileChannel.read(byteBuffer); byteBuffer.flip(); - CharBuffer chb = StandardCharsets.UTF_8.decode(byteBuffer); + CharBuffer charBuffer = StandardCharsets.UTF_8.decode(byteBuffer); byteBuffer.clear(); if (numRead <= 0) { throw new IOException("numRead less or equal to 0"); } - return chb.toString(); + return charBuffer.toString(); } - private static class LimitedMap extends LinkedHashMap { - - private final int n; - - LimitedMap(int n) { - super(); - this.n = n; + private void fillFileSizes(Path base, Pattern pattern) throws IOException { + if (!Files.exists(base)) { + return; } - - @Override - protected boolean removeEldestEntry(Map.Entry eldest) { - return size() > n; + try (Stream path = Files.walk(base)) { + path.forEach(p -> { + Matcher matcher = pattern.matcher(p.toString()); + if (matcher.matches()) { + try { + fileSizes.put(p, Files.size(p)); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + }); } } } diff --git a/src/test/java/org/xbib/event/io/file/FileFollowEventManagerTest.java b/src/test/java/org/xbib/event/io/file/FileFollowEventManagerTest.java index 695ebb4..13b86b2 100644 --- a/src/test/java/org/xbib/event/io/file/FileFollowEventManagerTest.java +++ b/src/test/java/org/xbib/event/io/file/FileFollowEventManagerTest.java @@ -6,27 +6,40 @@ import org.xbib.event.timer.TestTimerEventConsumer; import org.xbib.event.timer.TimerEventManager; import org.xbib.settings.Settings; +import java.io.BufferedWriter; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.time.Instant; import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; public class FileFollowEventManagerTest { + private static final Logger logger = Logger.getLogger(FileFollowEventManagerTest.class.getName()); + @Test public void testFileFollowEvents() throws IOException, InterruptedException { Path path = Files.createTempDirectory("testfilefollow"); TestFileFollowEventConsumer consumer = new TestFileFollowEventConsumer(); EventManager.register(consumer); Settings settings = Settings.settingsBuilder() - .put("event.timer.testfilefollowevent.enabled", "true") - .put("event.timer.testfilefollowevent.class", "org.xbib.event.io.file.TestFileFollowEvent") - .put("event.timer.testfilefollowevent.path", path.toString()) + .put("event.filefollow.testfilefollowevent.enabled", "true") + .put("event.filefollow.testfilefollowevent.class", "org.xbib.event.io.file.TestFileFollowEvent") + .put("event.filefollow.testfilefollowevent.base", path.toString()) + .put("event.filefollow.testfilefollowevent.pattern", ".*") .build(); EventManager eventManager = EventManager.newEventManager(settings); - FileFollowEventManager fileFolloeEventManager = eventManager.getFileFollowEventManager() - Thread.sleep(10000L); + FileFollowEventManager fileFolloeEventManager = eventManager.getFileFollowEventManager(); + Thread.sleep(5000L); + try (BufferedWriter bufferedWriter = Files.newBufferedWriter(path.resolve("test.txt"))) { + bufferedWriter.write("Hello"); + logger.log(Level.INFO, "Hello written"); + } + Thread.sleep(5000L); fileFolloeEventManager.close(); + Files.delete(path.resolve("test.txt")); + Files.delete(path); } } diff --git a/src/test/java/org/xbib/event/io/file/TestFileFollowEventConsumer.java b/src/test/java/org/xbib/event/io/file/TestFileFollowEventConsumer.java index e2f7738..3883ed3 100644 --- a/src/test/java/org/xbib/event/io/file/TestFileFollowEventConsumer.java +++ b/src/test/java/org/xbib/event/io/file/TestFileFollowEventConsumer.java @@ -16,6 +16,6 @@ public class TestFileFollowEventConsumer implements EventConsumer { @Subscribe @AllowConcurrentEvents void onEvent(TestFileFollowEvent event) { - logger.log(Level.INFO, "received filefollw event path = " + event.getPath()); + logger.log(Level.INFO, "received filefollw event path = " + event.getPath() + " content = " + event.getContent()); } }