INT-3989: Add FileReadingMessageSource.WatchServiceDirectoryScanner

JIRA: https://jira.spring.io/browse/INT-3989,
https://jira.spring.io/browse/INT-3990,
https://jira.spring.io/browse/INT-3988

* Deprecate top-level `WatchServiceDirectoryScanner` because of inconsistency around `Lifecycle` and shared `directory` property
* Copy/paste its logic into the `FileReadingMessageSource.WatchServiceDirectoryScanner` to hide that inconsistency, but still get a gain from the `WatchService` benefits
* Add support for the `StandardWatchEventKinds.ENTRY_MODIFY` and `StandardWatchEventKinds.ENTRY_DELETE` events in the `FileReadingMessageSource.WatchServiceDirectoryScanner`
* Introduce `useWatchService` option to switch to the internal `FileReadingMessageSource.WatchServiceDirectoryScanner`
* Make `CompositeFileListFilter` also as `ResettableFileListFilter`
* Deprecate weird `FileReadingMessageSource.onSend()` method and remove its usage from tests
* Modify `WatchServiceDirectoryScannerTests` for the new logic
* Document changes

Author: artembilan

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

@@ -17,6 +17,18 @@
 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.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.Comparator;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -27,11 +39,14 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.filters.ResettableFileListFilter;
+import org.springframework.lang.UsesJava7;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
@@ -64,7 +79,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  */
-public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource<File> {
+public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource<File>, Lifecycle {
 
 	private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5;
 
@@ -87,10 +102,14 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 
 	private volatile boolean scanEachPoll = false;
 
+	private volatile boolean running;
+
 	private FileListFilter<File> filter;
 
 	private FileLocker locker;
 
+	private boolean useWatchService;
+
 	/**
 	 * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity.
 	 */
@@ -236,11 +255,36 @@ public void setScanEachPoll(boolean scanEachPoll) {
 		this.scanEachPoll = scanEachPoll;
 	}
 
+	public void setUseWatchService(boolean useWatchService) {
+		this.useWatchService = useWatchService;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "file:inbound-channel-adapter";
 	}
 
+	@Override
+	public void start() {
+		if (this.scanner instanceof Lifecycle) {
+			((Lifecycle) this.scanner).start();
+		}
+		this.running = true;
+	}
+
+	@Override
+	public void stop() {
+		if (this.scanner instanceof Lifecycle) {
+			((Lifecycle) this.scanner).start();
+		}
+		this.running = false;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
 	@Override
 	protected void onInit() {
 		Assert.notNull(this.directory, "'directory' must not be null");
@@ -253,6 +297,14 @@ protected void onInit() {
 				"Source path [" + this.directory + "] does not point to a directory.");
 		Assert.isTrue(this.directory.canRead(),
 				"Source directory [" + this.directory + "] is not readable.");
+
+		Assert.state(!(this.scannerExplicitlySet && this.useWatchService),
+				"The 'scanner' and 'useWatchService' options are mutually exclusive: " + this.scanner);
+
+		if (this.useWatchService) {
+			this.scanner = new WatchServiceDirectoryScanner();
+		}
+
 		Assert.state(!(this.scannerExplicitlySet && (this.filter != null || this.locker != null)),
 				"The 'filter' and 'locker' options must be present on the provided external 'scanner': "
 						+ this.scanner);
@@ -262,6 +314,7 @@ protected void onInit() {
 		if (this.locker != null) {
 			this.scanner.setLocker(this.locker);
 		}
+
 	}
 
 	public Message<File> receive() throws MessagingException {
@@ -302,9 +355,7 @@ private void scanInputDirectory() {
 
 	/**
 	 * Adds the failed message back to the 'toBeReceived' queue if there is room.
-	 *
-	 * @param failedMessage
-	 *            the {@link org.springframework.messaging.Message} that failed
+	 * @param failedMessage the {@link Message} that failed
 	 */
 	public void onFailure(Message<File> failedMessage) {
 		if (logger.isWarnEnabled()) {
@@ -316,14 +367,149 @@ public void onFailure(Message<File> failedMessage) {
 	/**
 	 * The message is just logged. It was already removed from the queue during
 	 * the call to <code>receive()</code>
-	 *
-	 * @param sentMessage
-	 *            the message that was successfully delivered
+	 * @param sentMessage the message that was successfully delivered
+	 * @deprecated with no replacement. Redundant method.
 	 */
+	@Deprecated
 	public void onSend(Message<File> sentMessage) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Sent: " + sentMessage);
 		}
 	}
 
+	@UsesJava7
+	private class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements Lifecycle {
+
+		private volatile WatchService watcher;
+
+		private volatile Collection<File> initialFiles;
+
+		@Override
+		public void start() {
+				try {
+					this.watcher = FileSystems.getDefault().newWatchService();
+				}
+				catch (IOException e) {
+					logger.error("Failed to create watcher for " + FileReadingMessageSource.this.directory, e);
+				}
+				final Set<File> initialFiles = walkDirectory(FileReadingMessageSource.this.directory.toPath());
+				initialFiles.addAll(filesFromEvents());
+				this.initialFiles = initialFiles;
+		}
+
+		@Override
+		public void stop() {
+				try {
+					this.watcher.close();
+					this.watcher = null;
+				}
+				catch (IOException e) {
+					logger.error("Failed to close watcher for " + FileReadingMessageSource.this.directory, e);
+				}
+		}
+
+		@Override
+		public boolean isRunning() {
+			return true;
+		}
+
+		@Override
+		protected File[] listEligibleFiles(File directory) {
+			Assert.state(this.watcher != null, "The WatchService has'nt been 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 = this.watcher.poll();
+			Set<File> files = new LinkedHashSet<File>();
+			while (key != null) {
+				File parentDir = ((Path) key.watchable()).toAbsolutePath().toFile();
+				for (WatchEvent<?> event : key.pollEvents()) {
+					if (event.kind() == StandardWatchEventKinds.ENTRY_CREATE ||
+							event.kind() == StandardWatchEventKinds.ENTRY_MODIFY ||
+							event.kind() == StandardWatchEventKinds.ENTRY_DELETE) {
+						Path item = (Path) event.context();
+						File file = new File(parentDir, item.toFile().getName());
+						if (logger.isDebugEnabled()) {
+							logger.debug("Watch Event: " + event.kind() + ": " + file);
+						}
+
+						if (event.kind() == StandardWatchEventKinds.ENTRY_DELETE) {
+							if (FileReadingMessageSource.this.filter instanceof ResettableFileListFilter) {
+								((ResettableFileListFilter<File>) FileReadingMessageSource.this.filter).remove(file);
+							}
+						}
+						else {
+							if (file.isDirectory()) {
+								files.addAll(walkDirectory(file.toPath()));
+							}
+							else {
+								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(FileReadingMessageSource.this.directory.toPath()));
+						}
+					}
+				}
+				key.reset();
+				key = this.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 {
+						FileVisitResult fileVisitResult = super.preVisitDirectory(dir, attrs);
+						registerWatch(dir);
+						return fileVisitResult;
+					}
+
+					@Override
+					public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+						FileVisitResult fileVisitResult = super.visitFile(file, attrs);
+						walkedFiles.add(file.toFile());
+						return fileVisitResult;
+					}
+
+				});
+			}
+			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 events");
+			}
+			dir.register(this.watcher,
+					StandardWatchEventKinds.ENTRY_CREATE,
+					StandardWatchEventKinds.ENTRY_MODIFY,
+					StandardWatchEventKinds.ENTRY_DELETE);
+		}
+
+	}
+
 }

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

@@ -30,7 +30,7 @@
  * @author Iwein Fuld
  * @author Gary Russell
  *
- * @deprecated in favor of {@link WatchServiceDirectoryScanner} (when using Java 7 or later)
+ * @deprecated in favor of {@link FileReadingMessageSource#setUseWatchService(boolean)} (when using Java 7 or later)
  */
 @Deprecated
 public class RecursiveLeafOnlyDirectoryScanner extends DefaultDirectoryScanner {

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

@@ -60,9 +60,13 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @since 4.2
+ * @deprecated since 4.3 in favor of internal {@link WatchService} logic in the {@link FileReadingMessageSource}.
+ * Will be removed in Spring Integration 5.0.
  *
  */
+@Deprecated
 @UsesJava7
+@SuppressWarnings("deprecation")
 public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements SmartLifecycle {
 
 	private final static Log logger = LogFactory.getLog(WatchServiceDirectoryScanner.class);

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -46,6 +46,7 @@ protected BeanMetadataElement parseSource(Element element, ParserContext parserC
 				BeanDefinitionBuilder.genericBeanDefinition(FileReadingMessageSourceFactoryBean.class);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "comparator");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "scanner");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "use-watch-service");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size");

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java

@@ -54,6 +54,8 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean<FileRead
 
 	private volatile DirectoryScanner scanner;
 
+	private boolean useWatchService;
+
 	private volatile Boolean scanEachPoll;
 
 	private volatile Boolean autoCreateDirectory;
@@ -77,6 +79,10 @@ public void setScanner(DirectoryScanner scanner) {
 		this.scanner = scanner;
 	}
 
+	public void setUseWatchService(boolean useWatchService) {
+		this.useWatchService = useWatchService;
+	}
+
 	public void setFilter(FileListFilter<File> filter) {
 		if (filter instanceof AbstractFileLockerFilter && (this.locker == null)) {
 			this.setLocker((AbstractFileLockerFilter) filter);
@@ -146,6 +152,9 @@ else if (queueSizeSet) {
 			if (this.scanner != null) {
 				this.source.setScanner(this.scanner);
 			}
+			else {
+				this.source.setUseWatchService(this.useWatchService);
+			}
 			if (this.filter != null) {
 				if (this.locker == null) {
 					this.source.setFilter(this.filter);

spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -36,10 +36,11 @@
  * @author Iwein Fuld
  * @author Josh Long
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @param <F> The type that will be filtered.
  */
-public class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>, Closeable {
+public class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>, Closeable {
 
 	private final Set<FileListFilter<F>> fileFilters;
 
@@ -120,4 +121,16 @@ public void rollback(F file, List<F> files) {
 		}
 	}
 
+	@Override
+	public boolean remove(F f) {
+		boolean removed = false;
+		for (FileListFilter<F> fileFilter : this.fileFilters) {
+			if (fileFilter instanceof ResettableFileListFilter) {
+				((ResettableFileListFilter<F>) fileFilter).remove(f);
+				removed = true;
+			}
+		}
+		return removed;
+	}
+
 }