diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java index 5cc298ae7c..272f6c0f16 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java @@ -33,13 +33,13 @@ import java.util.Collection; import java.util.LinkedHashSet; import java.util.Set; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.context.SmartLifecycle; import org.springframework.lang.UsesJava7; import org.springframework.util.Assert; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - /** * Directory scanner that uses Java 7 {@link WatchService}. * @@ -49,11 +49,16 @@ import org.apache.commons.logging.LogFactory; * While initially walking the directory, any subdirectories encountered are registered * to watch for creation events. *

- * If subdirectories are subsequentially added, they too are walked and registered for - * new creation events. + * If subdirectories are subsequently added, they are walked and registered for + * new creation events, too. + *

+ * When a {@link StandardWatchEventKinds#OVERFLOW} {@link WatchKey} event is occurred, + * the {@link #directory} is rescanned to avoid the loss for any new entries according + * to the "missed events" logic around {@link StandardWatchEventKinds#OVERFLOW}. * * @author Hezi Schrager * @author Gary Russell + * @author Artem Bilan * @since 4.2 * */ @@ -175,8 +180,10 @@ public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implem for (WatchEvent event : key.pollEvents()) { if (event.kind() == StandardWatchEventKinds.ENTRY_CREATE) { Path item = (Path) event.context(); - File file = new File( - ((Path) key.watchable()).toAbsolutePath() + File.separator + item.getFileName()); + File file = new File(((Path) key.watchable()).toAbsolutePath() + File.separator + item.getFileName()); + if (logger.isDebugEnabled()) { + logger.debug("Watch Event: " + event.kind() + ": " + file); + } if (file.isDirectory()) { files.addAll(walkDirectory(file.toPath())); } @@ -184,6 +191,22 @@ public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implem files.add(file); } } + else if (event.kind() == StandardWatchEventKinds.OVERFLOW) { + if (logger.isDebugEnabled()) { + logger.debug("Watch Event: " + event.kind() + ": context: " + event.context()); + } + if (event.context() != null && event.context() instanceof Path) { + files.addAll(walkDirectory((Path) event.context())); + } + else { + files.addAll(walkDirectory(this.directory)); + } + } + else { + if (logger.isDebugEnabled()) { + logger.debug("Watch Event: " + event.kind() + ": context: " + event.context()); + } + } } key.reset(); key = watcher.poll(); @@ -199,14 +222,16 @@ public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implem @Override public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException { + FileVisitResult fileVisitResult = super.preVisitDirectory(dir, attrs); registerWatch(dir); - return super.preVisitDirectory(dir, attrs); + return fileVisitResult; } @Override public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException { + FileVisitResult fileVisitResult = super.visitFile(file, attrs); walkedFiles.add(file.toFile()); - return super.visitFile(file, attrs); + return fileVisitResult; } }); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java index 561955777e..372fee6664 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java @@ -22,7 +22,9 @@ import static org.junit.Assert.assertTrue; import java.io.File; import java.io.IOException; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; import org.junit.Before; import org.junit.Rule; @@ -31,6 +33,7 @@ import org.junit.rules.TemporaryFolder; /** * @author Gary Russell + * @author Artem Bilan * @since 4.2 * */ @@ -75,7 +78,7 @@ public class WatchServiceDirectoryScannerTests { File baz1 = File.createTempFile("baz", ".txt", baz); files = scanner.listFiles(folder.getRoot()); int n = 0; - List accum = new ArrayList(files); + Set accum = new HashSet(files); while (n++ < 300 && accum.size() != 4) { Thread.sleep(100); files = scanner.listFiles(folder.getRoot()); @@ -86,6 +89,42 @@ public class WatchServiceDirectoryScannerTests { assertTrue(accum.contains(foo2)); assertTrue(accum.contains(bar2)); assertTrue(accum.contains(baz1)); + + /*See AbstractWatchKey#signalEvent source code: + if(var5 >= 512) { + var1 = StandardWatchEventKinds.OVERFLOW; + } + */ + List filesForOverflow = new ArrayList(600); + + for (int i = 0; i < 600; i++) { + filesForOverflow.add(this.folder.newFile("" + i)); + } + + n = 0; + while (n++ < 300 && accum.size() < 604) { + Thread.sleep(100); + files = scanner.listFiles(folder.getRoot()); + accum.addAll(files); + } + + assertEquals(604, accum.size()); + + for (File fileForOverFlow : filesForOverflow) { + accum.contains(fileForOverFlow); + } + + File baz2 = File.createTempFile("baz2", ".txt", baz); + + n = 0; + while (n++ < 300 && accum.size() < 605) { + Thread.sleep(100); + files = scanner.listFiles(folder.getRoot()); + accum.addAll(files); + } + + assertTrue(accum.contains(baz2)); + scanner.stop(); } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 207125d129..5f009bedbd 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -217,6 +217,14 @@ On subsequent polls, files from new creation events are returned. If a new subdirectory is added, its creation event is used to walk the new subtree to find existing files, as well as registering any new subdirectories found. +NOTE: There is a case with `WatchKey`, when its internal events `queue` isn't drained by the program as quickly as +the directory modification events occur. +If the queue size is exceeded, a `StandardWatchEventKinds.OVERFLOW` is emitted to indicate that +some file system events may be lost. +In this case, the root directory is re-scanned completely. +To avoid duplicates consider using an appropriate `FileListFilter` such as the `AcceptOnceFileListFilter` and/or +remove files when processing is completed. + [source, xml] ----