INT-3229: Add WatchServiceDirectoryScanner

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

Polishing

spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,11 +26,16 @@ import java.util.List;
  * without limit. This scanner should not be used with directories that contain
  * a vast number of files or on deep trees, as all the file names will be read
  * into memory and the scanning will be done recursively.
- * 
+ *
  * @author Iwein Fuld
+ * @author Gary Russell
+ *
+ * @deprecated in favor of {@link WatchServiceDirectoryScanner} (when using Java 7 or later)
  */
+@Deprecated
 public class RecursiveLeafOnlyDirectoryScanner extends DefaultDirectoryScanner {
 
+	@Override
 	protected File[] listEligibleFiles(File directory) throws IllegalArgumentException {
 		File[] rootFiles = directory.listFiles();
 		List<File> files = new ArrayList<File>(rootFiles.length);

spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java

@@ -0,0 +1,227 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * 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.springframework.integration.file;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.StandardWatchEventKinds;
+import java.nio.file.WatchEvent;
+import java.nio.file.WatchKey;
+import java.nio.file.WatchService;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+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}.
+ *
+ * The initial state of the directory is collected during {@link #start()}. Subsequent
+ * polls return new files as reported by {@code ENTRY_CREATE} events.
+ * <p>
+ * 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.
+ *
+ * @author Hezi Schrager
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+@UsesJava7
+public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements SmartLifecycle {
+
+	private final static Log logger = LogFactory.getLog(WatchServiceDirectoryScanner.class);
+
+	private final Path directory;
+
+	private volatile WatchService watcher;
+
+	private volatile int phase;
+
+	private volatile boolean running;
+
+	private volatile boolean autoStartup;
+
+	private volatile Collection<File> initialFiles;
+
+	/**
+	 * Construct an instance for the given directory.
+	 * @param directory the directory.
+	 */
+	public WatchServiceDirectoryScanner(String directory) {
+		this.directory = Paths.get(directory);
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	/**
+	 * see {@link #getPhase()}
+	 * @param phase the phase.
+	 */
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	/**
+	 * @see #isRunning()
+	 * @param running true if running.
+	 */
+	public void setRunning(boolean running) {
+		this.running = running;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	/**
+	 * @see #isAutoStartup()
+	 * @param autoStartup true to auto start.
+	 */
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public synchronized void start() {
+		if (!this.running) {
+			try {
+				this.watcher = FileSystems.getDefault().newWatchService();
+			}
+			catch (IOException e) {
+				logger.error("Failed to create watcher for " + this.directory.toString(), e);
+			}
+			final Set<File> initialFiles = walkDirectory(this.directory);
+			initialFiles.addAll(filesFromEvents());
+			this.initialFiles = initialFiles;
+			this.running = true;
+		}
+	}
+
+	@Override
+	public synchronized void stop() {
+		if (this.running) {
+			try {
+				this.watcher.close();
+			}
+			catch (IOException e) {
+				logger.error("Failed to close watcher for " + this.directory.toString(), e);
+			}
+			this.running = false;
+		}
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		stop();
+		callback.run();
+	}
+
+	@Override
+	protected File[] listEligibleFiles(File directory) {
+		Assert.state(this.watcher != null, "Scanner needs to be started");
+		if (this.initialFiles != null) {
+			File[] initial = this.initialFiles.toArray(new File[this.initialFiles.size()]);
+			this.initialFiles = null;
+			return initial;
+		}
+		Collection<File> files = filesFromEvents();
+		return files.toArray(new File[files.size()]);
+	}
+
+	private Set<File> filesFromEvents() {
+		WatchKey key = watcher.poll();
+		Set<File> files = new LinkedHashSet<File>();
+		while (key != null) {
+			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());
+					if (file.isDirectory()) {
+						files.addAll(walkDirectory(file.toPath()));
+					}
+					else {
+						files.add(file);
+					}
+				}
+			}
+			key.reset();
+			key = watcher.poll();
+		}
+		return files;
+	}
+
+	private Set<File> walkDirectory(Path directory) {
+		final Set<File> walkedFiles = new LinkedHashSet<File>();
+		try {
+			registerWatch(directory);
+			Files.walkFileTree(directory, new SimpleFileVisitor<Path>() {
+
+				@Override
+				public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
+					registerWatch(dir);
+					return super.preVisitDirectory(dir, attrs);
+				}
+
+				@Override
+				public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+					walkedFiles.add(file.toFile());
+					return super.visitFile(file, attrs);
+				}
+
+			});
+		}
+		catch (IOException e) {
+			logger.error("Failed to walk directory: " + directory.toString(), e);
+		}
+		return walkedFiles;
+	}
+
+	private void registerWatch(Path dir) throws IOException {
+		if (logger.isDebugEnabled()) {
+			logger.debug("registering: " + dir + " for file creation events");
+		}
+		dir.register(this.watcher, StandardWatchEventKinds.ENTRY_CREATE);
+	}
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScannerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,9 +16,9 @@
 
 package org.springframework.integration.file;
 
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertEquals;
 import static org.hamcrest.CoreMatchers.hasItem;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 
 import java.io.File;
 import java.io.IOException;
@@ -31,6 +31,7 @@ import org.junit.rules.TemporaryFolder;
 /**
  * @author Iwein Fuld
  * @author Gunnar Hillert
+ * @author Gary Russell
  */
 public class RecursiveLeafOnlyDirectoryScannerTests {
 
@@ -65,6 +66,7 @@ public class RecursiveLeafOnlyDirectoryScannerTests {
 
 	@Test
 	public void shouldReturnAllFiles() {
+		@SuppressWarnings("deprecation")
 		List<File> files = new RecursiveLeafOnlyDirectoryScanner().listFiles(recursivePath.getRoot());
 		assertEquals(Integer.valueOf(files.size()), Integer.valueOf(3));
 		assertThat(files, hasItem(topLevelFile));

spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * 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.springframework.integration.file;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class WatchServiceDirectoryScannerTests {
+
+	@Rule
+	public TemporaryFolder folder = new TemporaryFolder();
+
+	private File foo;
+
+	private File bar;
+
+	private File top1;
+
+	private File foo1;
+
+	private File bar1;
+
+	@Before
+	public void setUp() throws IOException {
+		this.foo = this.folder.newFolder("foo");
+		this.bar = this.folder.newFolder("bar");
+		this.top1 = this.folder.newFile();
+		this.foo1 = File.createTempFile("foo", ".txt", this.foo);
+		this.bar1 = File.createTempFile("bar", ".txt", this.bar);
+	}
+
+	@Test
+	public void testInitial() throws Exception {
+		WatchServiceDirectoryScanner scanner = new WatchServiceDirectoryScanner(folder.getRoot().getAbsolutePath());
+		scanner.start();
+		List<File> files = scanner.listFiles(folder.getRoot());
+		assertEquals(3, files.size());
+		assertTrue(files.contains(top1));
+		assertTrue(files.contains(foo1));
+		assertTrue(files.contains(bar1));
+		File top2 = this.folder.newFile();
+		File foo2 = File.createTempFile("foo", ".txt", this.foo);
+		File bar2 = File.createTempFile("bar", ".txt", this.bar);
+		File baz = new File(this.foo, "baz");
+		baz.mkdir();
+		File baz1 = File.createTempFile("baz", ".txt", baz);
+		files = scanner.listFiles(folder.getRoot());
+		int n = 0;
+		while (n++ < 200 && files.size() == 0) {
+			Thread.sleep(100);
+			files = scanner.listFiles(folder.getRoot());
+		}
+		assertEquals(4, files.size());
+		assertTrue(files.contains(top2));
+		assertTrue(files.contains(foo2));
+		assertTrue(files.contains(bar2));
+		assertTrue(files.contains(baz1));
+		scanner.stop();
+	}
+
+}

src/reference/asciidoc/file.adoc

@@ -204,6 +204,34 @@ IMPORTANT: It is important to understand that filters (including patterns, regex
 Any of these attributes set on the adapter are subsequently injected into the scanner.
 For this reason, if you need to provide a custom scanner and you have multiple file inbound adapters in the same application context, each adapter must be provided with its own instance of the scanner, either by declaring separate beans, or declaring `scope="prototype"` on the scanner bean so that the context will create a new instance for each use.
 
+==== WatchServiceDirectoryScanner
+
+This scanner was added in _version 4.2_. It replaces the existing `RecursiveLeafOnlyDirectoryScanner` which is
+inefficient for large directory trees. The `WatchServiceDirectoryScanner` requires Java 7 or above.
+
+This scanner relies on file system events when new files are added to the directory.
+During initialization, the directory is registered to generate events; the initial file list is also built.
+While walking the directory tree, any subdirectories encountered are also registered to generate events.
+On the first poll, the initial file list from walking the directory is returned.
+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.
+
+[source, xml]
+----
+<bean id="wsScanner" class="org.springframework.integration.file.WatchServiceDirectoryScanner">
+    <constructor-arg value="/tmp/myDir" />
+</bean>
+----
+
+[source, java]
+----
+@Bean
+public DirectoryScanner scanner() {
+    return new WatchServiceDirectoryScanner("/tmp/myDir");
+}
+----
+
 ==== Limiting Memory Consumption
 
 A `HeadDirectoryScanner` can be used to limit the number of files retained in memory.

src/reference/asciidoc/whats-new.adoc

@@ -84,6 +84,10 @@ The `HeadDirectoryScanner` can now be used with other `FileListFilter` s.
 
 The `LastModifiedFileListFilter` has been added.
 
+===== WatchService Directory Scanner
+
+The `WatchServiceDirectoryScanner` is now available.
+
 [[x4.2-class-package-change]]
 ==== Class Package Change