Add DirectoryListener

metafacture-io/src/main/java/org/metafacture/io/DirectoryListener.java

@@ -0,0 +1,206 @@
+/*
+ * Copyright 2026, hbz
+ *
+ * Licensed under the Apache License, Version 2.0 the "License";
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.metafacture.io;
+
+import org.metafacture.framework.FluxCommand;
+import org.metafacture.framework.MetafactureLogger;
+import org.metafacture.framework.ObjectReceiver;
+import org.metafacture.framework.annotations.Description;
+import org.metafacture.framework.annotations.In;
+import org.metafacture.framework.annotations.Out;
+import org.metafacture.framework.helpers.DefaultObjectPipe;
+
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.WatchEvent;
+import java.nio.file.WatchKey;
+import java.nio.file.WatchService;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Listens to a directory and passes occurring filenames to the receiver.
+ * If a file named {@value TRIGGER_SHUTDOWN_FILENAME} appears the process
+ * is closed.
+ * Keep bug @see <a href="https://bugs.openjdk.org/browse/JDK-8202759">JDK-8202759</a>
+ * in mind: if files occur too fast the files may be missed by the watcher.
+ *
+ * @author Pascal Christoph (dr0i)
+ */
+@Description("Listens to a directory and passes filenames of occurring or modified files to the receiver." +
+        "If a file named 'shutdownEtlNow' appears the process " +
+        "is closed." +
+        "Keep bug https://bugs.openjdk.org/browse/JDK-8202759 " +
+        "in mind: if files occur too fast the files may be missed by the watcher.")
+@In(String.class)
+@Out(String.class)
+@FluxCommand("listen-directory")
+public final class DirectoryListener extends DefaultObjectPipe<String, ObjectReceiver<String>> {
+
+    /* This special filename triggers the end of listing and closes the module */
+    public static final String TRIGGER_SHUTDOWN_FILENAME = "shutdownEtlNow";
+    private static final WatchService WATCHER;
+    private static final MetafactureLogger LOG = new MetafactureLogger(DirectoryListener.class);
+
+    static {
+        try {
+            WATCHER = FileSystems.getDefault().newWatchService();
+        }
+        catch (final IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static final Map<WatchKey, Path> KEYS = new HashMap<>();
+
+    /**
+     * Creates an instance of {@link DirectoryListener} if no IOException occurs.
+     */
+    public DirectoryListener() {
+    }
+
+    @Override
+    public void process(final String directory) {
+
+        final Path dir = Path.of(directory);
+        try {
+            registerAll(dir);
+        }
+        catch (final IOException e) {
+            throw new RuntimeException(e);
+        }
+        start(directory);
+    }
+
+    private void start(final String directory) {
+        final DirectoryWatcher directoryWatcher = new DirectoryWatcher();
+        directoryWatcher.setDirectory(directory);
+        final Thread thread = new Thread(directoryWatcher);
+        thread.start();
+    }
+
+    /**
+     * Register the given directory with the WatchService.
+     *
+     * @param dir the directory to register
+     */
+    private void register(final Path dir) throws IOException {
+        final WatchKey key = dir.register(WATCHER, java.nio.file.StandardWatchEventKinds.ENTRY_CREATE, java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY);
+        LOG.combinedInfo("Add directory to watch: " + dir.toString());
+        KEYS.put(key, dir);
+    }
+
+    /**
+     * Register the given directory, and all its subdirectories, with the
+     * WatchService.
+     *
+     * @param start root directory for registering all (sub)directories
+     */
+    private void registerAll(final Path start) throws IOException {
+        Files.walkFileTree(start, new SimpleFileVisitor<>() {
+            @Override
+            public FileVisitResult preVisitDirectory(final Path dir, final BasicFileAttributes attrs)
+                    throws IOException {
+                register(dir);
+                return FileVisitResult.CONTINUE;
+            }
+        });
+    }
+
+    final class DirectoryWatcher implements Runnable {
+        private String directory;
+
+        DirectoryWatcher() {
+        }
+
+        private void setDirectory(final String directory) {
+            this.directory = directory;
+        }
+
+        public void run() {
+
+            while (true) {
+                final WatchKey key;
+                try {
+                    key = WATCHER.take();
+                }
+                catch (final InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    return;
+                }
+                final Path dir = KEYS.get(key);
+                if (dir == null) {
+                    LOG.warn("WatchKey not recognized!");
+                    continue;
+                }
+
+                for (final WatchEvent<?> event : key.pollEvents()) {
+                    // an OVERFLOW event can occur if events are lost or discarded
+                    if (event.kind() == java.nio.file.StandardWatchEventKinds.OVERFLOW) {
+                        throw new OpenFailed("Overflow event occurred on directory " + directory);
+                    }
+                    LOG.info("Event kind {} on file: '{}'", event.kind(),  event.context());
+
+                    @SuppressWarnings("unchecked")
+                    final Path fileName = ((WatchEvent<Path>) event).context();
+                    final Path absolutePath = dir.resolve(fileName);
+                    processFile(fileName, absolutePath);
+                }
+                // reset key and remove from set if directory no longer accessible
+                final boolean valid = key.reset();
+                if (!valid) {
+                    KEYS.remove(key);
+                    LOG.info("Directory no longer accessible: " + key.toString());
+                    // all directories are inaccessible
+                    if (KEYS.isEmpty()) {
+                        LOG.combinedWarn("Root directory {} is not accessible anymore. Closing ...", directory);
+                        closeStream();
+                        break;
+                    }
+                }
+            }
+        }
+
+        private void processFile(final Path fileName, final Path absolutePath) {
+            if (Files.isDirectory(absolutePath, LinkOption.NOFOLLOW_LINKS)) {
+                try {
+                    registerAll(absolutePath);
+                }
+                catch (final IOException e) {
+                    throw new OpenFailed("IOException event occurred on directory " + directory, e);
+                }
+            }
+            else {
+                if (fileName.toString().equals(TRIGGER_SHUTDOWN_FILENAME)) {
+                    LOG.combinedInfo("Shutdown triggered. Going down ...");
+                    closeStream();
+                    Thread.currentThread().interrupt();
+                }
+                else {
+                    LOG.combinedDebug("processing '{}'", absolutePath.toString());
+                    getReceiver().process(absolutePath.toString());
+                }
+            }
+        }
+    }
+}

metafacture-io/src/main/resources/flux-commands.properties

@@ -23,3 +23,4 @@ as-records org.metafacture.io.RecordReader
 open-resource org.metafacture.io.ResourceOpener
 open-tar org.metafacture.io.TarReader
 open-sru org.metafacture.io.SruOpener
+listen-directory org.metafacture.io.DirectoryListener

metafacture-io/src/test/java/org/metafacture/io/DirectoryListenerTest.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright 2026, hbz
+ *
+ * Licensed under the Apache License, Version 2.0 the "License";
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.metafacture.io;
+
+import org.metafacture.framework.ObjectReceiver;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+
+import static org.mockito.Mockito.times;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * Tests for class {@link DirectoryListener}.
+ *
+ * @author Pascal Christoph (dr0i)
+ */
+public final class DirectoryListenerTest {
+
+    private static final DirectoryListener DIRECTORY_LISTENER = new DirectoryListener();
+    private static final int MAX_MILLISECONDS_WAITING_OF_THREAD = 3000;
+    private static final String FILE_NAME = "test";
+    private static final String SUBDIRECTORY_NAME = "subdir";
+
+    @Rule
+    public MockitoRule mockitoRule = MockitoJUnit.rule();
+
+    @Rule
+    public TemporaryFolder tempFolder = new TemporaryFolder();
+
+    private String pathToDirectory;
+    private String pathToSubdirectory;
+
+    @Mock
+    private ObjectReceiver<String> receiver;
+
+    public DirectoryListenerTest() {
+    }
+
+    @Before
+    public void setup() {
+        pathToDirectory = tempFolder.getRoot() + File.separator;
+        DIRECTORY_LISTENER.setReceiver(receiver);
+        DIRECTORY_LISTENER.process(pathToDirectory);
+        pathToSubdirectory = pathToDirectory + SUBDIRECTORY_NAME + File.separator;
+    }
+
+    @Test
+    public void testFileOccurs() {
+        final String pathToTestfile = pathToDirectory + FILE_NAME;
+        createFile(pathToTestfile);
+        Mockito.verify(receiver, org.mockito.Mockito.timeout(MAX_MILLISECONDS_WAITING_OF_THREAD)).process(pathToTestfile);
+    }
+
+    @Test
+    public void testFileOccursInSubdirectory() throws InterruptedException {
+        createDirectory(pathToSubdirectory);
+        final String pathToTestfile = pathToSubdirectory + FILE_NAME;
+        Thread.sleep(100); // because of https://bugs.openjdk.org/browse/JDK-8202759
+        createFile(pathToTestfile);
+        Mockito.verify(receiver, org.mockito.Mockito.timeout(MAX_MILLISECONDS_WAITING_OF_THREAD)).process(pathToTestfile);
+    }
+
+        @Test
+    public void testFileOccursAndIsIgnoredWhenDeleted() throws InterruptedException {
+        final String pathToTestfile = pathToDirectory + FILE_NAME + "1";
+        createFile(pathToTestfile);
+        Mockito.verify(receiver, org.mockito.Mockito.timeout(MAX_MILLISECONDS_WAITING_OF_THREAD)).process(pathToTestfile);
+        Mockito.verify(receiver,times(1)).process(pathToTestfile);
+        removeFile(pathToTestfile);
+        Thread.sleep(100); // because of https://bugs.openjdk.org/browse/JDK-8202759
+        Mockito.verify(receiver,times(1)).process(pathToTestfile);
+
+    }
+
+    @Test
+    public void testDontProcessDirectoryWithoutFiles() throws InterruptedException {
+        final String pathToTestfile = pathToSubdirectory;
+        createFile(pathToTestfile);
+        Thread.sleep(100); // because of https://bugs.openjdk.org/browse/JDK-8202759
+        Mockito.verify(receiver, org.mockito.Mockito.timeout(MAX_MILLISECONDS_WAITING_OF_THREAD).times(0)).process(pathToTestfile);
+    }
+
+    @Test
+    public void testTriggerShutdown() throws InterruptedException {
+        final String pathToTestfile = pathToDirectory + DirectoryListener.TRIGGER_SHUTDOWN_FILENAME;
+        createFile(pathToTestfile);
+        Mockito.verify(receiver, org.mockito.Mockito.timeout(MAX_MILLISECONDS_WAITING_OF_THREAD).times(0)).process(
+                pathToDirectory);
+        Thread.sleep(100);
+        Assert.assertTrue(DIRECTORY_LISTENER.isClosed());
+    }
+
+    private void createFile(final String path) {
+        final File testFile = new File(path);
+        try {
+            testFile.createNewFile();
+        }
+        catch (final IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void createDirectory(final String dir) {
+        try {
+            Files.createDirectory(Path.of(dir));
+        }
+        catch (final IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void removeFile(final String path) {
+        final File testFile = new File(path);
+        testFile.delete();
+    }
+}

metafacture-runner/src/main/dist/examples/misc/directoryListener/directoryListener.flux

@@ -0,0 +1,5 @@
+default infile = FLUX_DIR + "tmp";
+infile|
+listen-directory|
+write (FLUX_DIR + "result.txt")
+;

metafacture-runner/src/main/dist/examples/misc/directoryListener/tmp/emptyFile.txt

@@ -0,0 +1 @@
+hello, new