INT-3804: WatchServiceDirScan: Add OVERFLOW Event

JIRA: https://jira.spring.io/browse/INT-3804

INT-3804: Polishing

Fix race conditions.

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.
  * <p>
- * 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.
+ * <p>
+ * 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;
 				}
 
 			});

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<File> accum = new ArrayList<File>(files);
+		Set<File> accum = new HashSet<File>(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<File> filesForOverflow = new ArrayList<File>(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();
 	}
 

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]
 ----
 <bean id="wsScanner" class="org.springframework.integration.file.WatchServiceDirectoryScanner">