Use MetafactureLogger (#702)

- improve doc
- test removing a file

Author: dr0i

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

@@ -17,6 +17,7 @@
 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;
@@ -46,7 +47,7 @@
  *
  * @author Pascal Christoph (dr0i)
  */
-@Description("Listens to a directory and passes occurring filenames to the receiver. " +
+@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 " +
@@ -59,6 +60,7 @@ public final class DirectoryListener extends DefaultObjectPipe<String, ObjectRec
     /* 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 {
@@ -104,7 +106,7 @@ private void start(final String directory) {
      */
     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);
-        System.out.println("Add directory to watch: " + dir);
+        LOG.combinedInfo("Add directory to watch: " + dir.toString());
         KEYS.put(key, dir);
     }
 
@@ -148,7 +150,7 @@ public void run() {
                 }
                 final Path dir = KEYS.get(key);
                 if (dir == null) {
-                    System.err.println("WatchKey not recognized!");
+                    LOG.warn("WatchKey not recognized!");
                     continue;
                 }
 
@@ -157,20 +159,22 @@ public void run() {
                     if (event.kind() == java.nio.file.StandardWatchEventKinds.OVERFLOW) {
                         throw new OpenFailed("Overflow event occurred on directory " + directory);
                     }
-                    System.out.println("Event kind:" + event.kind() + ". File affected: " + event.context() + ".");
+                    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;
                     }
                 }
@@ -188,10 +192,12 @@ private void processFile(final Path fileName, final Path absolutePath) {
             }
             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/test/java/org/metafacture/io/DirectoryListenerTest.java

@@ -28,6 +28,8 @@
 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;
@@ -84,11 +86,23 @@ public void testFileOccursInSubdirectory() throws InterruptedException {
         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 = pathToDirectory + SUBDIRECTORY_NAME;
-        Thread.sleep(400); // because of https://bugs.openjdk.org/browse/JDK-8202759
+        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);
     }
 
@@ -120,4 +134,9 @@ private void createDirectory(final String dir) {
             throw new RuntimeException(e);
         }
     }
+
+    private void removeFile(final String path) {
+        final File testFile = new File(path);
+        testFile.delete();
+    }
 }