INT-4334: Adds a DirectoryScanner to RemoteFileMS

JIRA: https://jira.springsource.org/browse/INT-4334

* Add a `DirectoryScanner` setter named `setScanner` in
`AbstractInboundFileSynchronizingMessageSource` to set the
`FileReadingMessageSource`'s scanner.
* Add tests to `AbstractRemoteFileSynchronizerTests`.

Polish.

Updates xsd, tests and documentation.

Fix xsd formatting.

Addresses review comments.

* Some code style polishing

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

@@ -37,6 +37,7 @@ import org.springframework.util.StringUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Venil Noronha
  *
  * @since 2.0
  */
@@ -75,6 +76,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 			messageSourceBuilder.addConstructorArgReference(comparator);
 		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(messageSourceBuilder, element, "local-filter");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(messageSourceBuilder, element, "scanner");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element,
 				"auto-create-local-directory");

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java

@@ -26,6 +26,7 @@ import java.util.regex.Pattern;
 import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource;
+import org.springframework.integration.file.DirectoryScanner;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.RecursiveDirectoryScanner;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
@@ -59,6 +60,7 @@ import org.springframework.util.Assert;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Venil Noronha
  */
 public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 		extends AbstractFetchLimitingMessageSource<File> implements Lifecycle {
@@ -88,6 +90,10 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 
 	private volatile FileListFilter<File> localFileListFilter;
 
+	/**
+	 * Whether the {@link DirectoryScanner} was explicitly set.
+	 */
+	private volatile boolean scannerExplicitlySet = false;
 
 	public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer) {
 		this(synchronizer, null);
@@ -145,6 +151,17 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 		}
 	}
 
+	/**
+	 * Switch the local {@link FileReadingMessageSource} to use a custom
+	 * {@link DirectoryScanner}.
+	 * @param scanner the {@link DirectoryScanner} to use.
+	 * @since 5.0
+	 */
+	public void setScanner(DirectoryScanner scanner) {
+		this.fileSource.setScanner(scanner);
+		this.scannerExplicitlySet = true;
+	}
+
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		super.afterPropertiesSet();
@@ -167,7 +184,12 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 						new SimpleMetadataStore(), getComponentName());
 			}
 			FileListFilter<File> filter = buildFilter();
-			if (!this.fileSource.isUseWatchService()) {
+			if (this.scannerExplicitlySet) {
+				Assert.state(!this.fileSource.isUseWatchService(),
+						"'useWatchService' and 'scanner' are mutually exclusive.");
+				this.fileSource.getScanner().setFilter(filter);
+			}
+			else if (!this.fileSource.isUseWatchService()) {
 				RecursiveDirectoryScanner directoryScanner = new RecursiveDirectoryScanner();
 				directoryScanner.setFilter(filter);
 				this.fileSource.setScanner(directoryScanner);

spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java

@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.file.HeadDirectoryScanner;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
@@ -41,6 +42,7 @@ import org.springframework.messaging.MessagingException;
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Venil Noronha
  *
  * @since 4.0.4
  *
@@ -116,21 +118,7 @@ public class AbstractRemoteFileSynchronizerTests {
 	public void testMaxFetchSizeSource() throws Exception {
 		final AtomicInteger count = new AtomicInteger();
 		AbstractInboundFileSynchronizer<String> sync = createLimitingSynchronizer(count);
-
+		AbstractInboundFileSynchronizingMessageSource<String> source = createSource(sync);
-		AbstractInboundFileSynchronizingMessageSource<String> source =
-				new AbstractInboundFileSynchronizingMessageSource<String>(sync) {
-
-			@Override
-			public String getComponentType() {
-				return "foo";
-			}
-
-		};
-		source.setLocalDirectory(new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID()));
-		source.setAutoCreateLocalDirectory(true);
-		source.setBeanFactory(mock(BeanFactory.class));
-		source.setMaxFetchSize(1);
-		source.setBeanName("maxFetchSizeSource");
 		source.afterPropertiesSet();
 		source.start();
 
@@ -143,6 +131,60 @@ public class AbstractRemoteFileSynchronizerTests {
 		source.stop();
 	}
 
+	@Test
+	public void testExclusiveScanner() throws Exception {
+		final AtomicInteger count = new AtomicInteger();
+		AbstractInboundFileSynchronizingMessageSource<String> source = createSource(count);
+		source.setScanner(new HeadDirectoryScanner(1));
+		source.afterPropertiesSet();
+		source.start();
+		source.receive();
+		assertEquals(1, count.get());
+	}
+
+	@Test
+	public void testExclusiveWatchService() throws Exception {
+		final AtomicInteger count = new AtomicInteger();
+		AbstractInboundFileSynchronizingMessageSource<String> source = createSource(count);
+		source.setUseWatchService(true);
+		source.afterPropertiesSet();
+		source.start();
+		source.receive();
+		assertEquals(1, count.get());
+	}
+
+	@Test(expected = IllegalStateException.class)
+	public void testScannerAndWatchServiceConflict() throws Exception {
+		final AtomicInteger count = new AtomicInteger();
+		AbstractInboundFileSynchronizingMessageSource<String> source = createSource(count);
+		source.setUseWatchService(true);
+		source.setScanner(new HeadDirectoryScanner(1));
+		source.afterPropertiesSet();
+	}
+
+	private AbstractInboundFileSynchronizingMessageSource<String> createSource(AtomicInteger count) {
+		return createSource(createLimitingSynchronizer(count));
+	}
+
+	private AbstractInboundFileSynchronizingMessageSource<String> createSource(
+			AbstractInboundFileSynchronizer<String> sync) {
+		AbstractInboundFileSynchronizingMessageSource<String> source =
+				new AbstractInboundFileSynchronizingMessageSource<String>(sync) {
+
+					@Override
+					public String getComponentType() {
+						return "foo";
+					}
+
+				};
+		source.setMaxFetchSize(1);
+		source.setLocalDirectory(new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID()));
+		source.setAutoCreateLocalDirectory(true);
+		source.setBeanFactory(mock(BeanFactory.class));
+		source.setBeanName("fooSource");
+		return source;
+	}
+
 	private AbstractInboundFileSynchronizer<String> createLimitingSynchronizer(final AtomicInteger count) {
 		SessionFactory<String> sf = new StringSessionFactory();
 		AbstractInboundFileSynchronizer<String> sync = new AbstractInboundFileSynchronizer<String>(sf) {
@@ -194,7 +236,7 @@ public class AbstractRemoteFileSynchronizerTests {
 
 		@Override
 		public String[] list(String path) throws IOException {
-			return new String[] {"foo", "bar", "baz"};
+			return new String[] { "foo", "bar", "baz" };
 		}
 
 		@Override

spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd

@@ -163,6 +163,19 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="scanner" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.integration.file.DirectoryScanner" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								Reference to a custom DirectoryScanner implementation.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attributeGroup ref="tempSuffixGroup" />
 				</xsd:extension>
 			</xsd:complexContent>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml

@@ -17,6 +17,7 @@
 	<int-ftp:inbound-channel-adapter id="ftpInbound"
 				channel="ftpChannel"
 				session-factory="ftpSessionFactory"
+				scanner="dirScanner"
 				auto-create-local-directory="true"
 				auto-startup="false"
 				delete-remote-files="true"
@@ -35,6 +36,10 @@
 			</int:poller>
 	</int-ftp:inbound-channel-adapter>
 
+	<bean id="dirScanner" class="org.springframework.integration.file.HeadDirectoryScanner">
+		<constructor-arg value="1" />
+	</bean>
+
 	<bean id="fooString" class="java.lang.String">
 		<constructor-arg value="foo" />
 	</bean>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java

@@ -43,6 +43,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.DirectoryScanner;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
@@ -66,6 +67,7 @@ import org.springframework.util.ReflectionUtils;
  * @author Gary Russell
  * @author Gunnar Hillert
  * @author Artem Bilan
+ * @author Venil Noronha
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -88,20 +90,26 @@ public class FtpInboundChannelAdapterParserTests {
 	@Autowired
 	private ApplicationContext context;
 
+	@Autowired
+	private DirectoryScanner dirScanner;
+
 	@Test
 	public void testFtpInboundChannelAdapterComplete() throws Exception {
 		assertFalse(TestUtils.getPropertyValue(ftpInbound, "autoStartup", Boolean.class));
-		PriorityBlockingQueue<?> blockingQueue = TestUtils.getPropertyValue(ftpInbound, "source.fileSource.toBeReceived", PriorityBlockingQueue.class);
+		PriorityBlockingQueue<?> blockingQueue =
+				TestUtils.getPropertyValue(ftpInbound, "source.fileSource.toBeReceived", PriorityBlockingQueue.class);
 		Comparator<?> comparator = blockingQueue.comparator();
 		assertNotNull(comparator);
 		assertEquals("ftpInbound", ftpInbound.getComponentName());
 		assertEquals("ftp:inbound-channel-adapter", ftpInbound.getComponentType());
 		assertEquals(context.getBean("ftpChannel"), TestUtils.getPropertyValue(ftpInbound, "outputChannel"));
 		FtpInboundFileSynchronizingMessageSource inbound =
-			(FtpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(ftpInbound, "source");
+				(FtpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(ftpInbound, "source");
+
+		assertSame(dirScanner, TestUtils.getPropertyValue(inbound, "fileSource.scanner"));
 
 		FtpInboundFileSynchronizer fisync =
-			(FtpInboundFileSynchronizer) TestUtils.getPropertyValue(inbound, "synchronizer");
+				(FtpInboundFileSynchronizer) TestUtils.getPropertyValue(inbound, "synchronizer");
 		assertEquals("'foo/bar'", TestUtils.getPropertyValue(fisync, "remoteDirectoryExpression", Expression.class)
 				.getExpressionString());
 		assertNotNull(TestUtils.getPropertyValue(fisync, "localFilenameGeneratorExpression"));
@@ -123,7 +131,8 @@ public class FtpInboundChannelAdapterParserTests {
 		Object sessionFactory = TestUtils.getPropertyValue(fisync, "remoteFileTemplate.sessionFactory");
 		assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass()));
 		FileListFilter<?> acceptAllFilter = context.getBean("acceptAllFilter", FileListFilter.class);
-		assertTrue(TestUtils.getPropertyValue(inbound, "fileSource.scanner.filter.fileFilters", Collection.class).contains(acceptAllFilter));
+		assertTrue(TestUtils.getPropertyValue(inbound, "fileSource.scanner.filter.fileFilters", Collection.class)
+				.contains(acceptAllFilter));
 		final AtomicReference<Method> genMethod = new AtomicReference<Method>();
 		ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class, method -> {
 			method.setAccessible(true);
@@ -135,16 +144,19 @@ public class FtpInboundChannelAdapterParserTests {
 
 	@Test
 	public void cachingSessionFactory() throws Exception {
-		Object sessionFactory = TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.synchronizer.remoteFileTemplate.sessionFactory");
+		Object sessionFactory = TestUtils.getPropertyValue(simpleAdapterWithCachedSessions,
+				"source.synchronizer.remoteFileTemplate.sessionFactory");
 		assertEquals(CachingSessionFactory.class, sessionFactory.getClass());
 		FtpInboundFileSynchronizer fisync =
-			TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.synchronizer", FtpInboundFileSynchronizer.class);
+				TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.synchronizer",
+						FtpInboundFileSynchronizer.class);
 		String remoteFileSeparator = (String) TestUtils.getPropertyValue(fisync, "remoteFileSeparator");
 		assertNotNull(remoteFileSeparator);
 		assertEquals("/", remoteFileSeparator);
 		assertEquals("foo/bar", TestUtils.getPropertyValue(fisync, "remoteDirectoryExpression", Expression.class)
 				.getExpressionString());
-		assertEquals(Integer.MIN_VALUE, TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.maxFetchSize"));
+		assertEquals(Integer.MIN_VALUE,
+				TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.maxFetchSize"));
 	}
 
 	@Test
@@ -171,6 +183,7 @@ public class FtpInboundChannelAdapterParserTests {
 		public boolean isSingleton() {
 			return true;
 		}
+
 	}
 
 }

spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd

@@ -167,6 +167,19 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="scanner" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.integration.file.DirectoryScanner" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								Reference to a custom DirectoryScanner implementation.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attributeGroup ref="tempSuffixGroup" />
 				</xsd:extension>
 			</xsd:complexContent>

src/reference/asciidoc/ftp.adoc

@@ -240,6 +240,7 @@ The _FTP Inbound Channel Adapter_ is a special listener that will connect to the
     remote-file-separator="/"
     preserve-timestamp="true"
     local-filename-generator-expression="#this.toUpperCase() + '.a'"
+    scanner="myDirScanner"
     local-filter="myFilter"
     temporary-file-suffix=".writing"
     max-fetch-size="-1"
@@ -259,6 +260,7 @@ The inbound channel adapter first retrieves the file to a local directory and th
 Starting with _version 5.0_, you can now limit the number of files fetched from the FTP server when new file retrievals are needed.
 This can be beneficial when the target files are very large and/or when running in a clustered system with a persistent file list filter discussed below.
 Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved; see <<ftp-max-fetch>> for more information.
+Since _version 5.0_, you can also provide a custom `DirectoryScanner` implementation to the `inbound-channel-adapter` via the `scanner` attribute.
 
 Starting with _Spring Integration 3.0_, you can specify the `preserve-timestamp` attribute (default `false`); when `true`, the local file's modified timestamp will be set to the value retrieved from the server; otherwise it will be set to the current time.
 

src/reference/asciidoc/sftp.adoc

@@ -311,6 +311,7 @@ The _SFTP Inbound Channel Adapter_ is a special listener that will connect to th
             local-directory="file:target/foo"
             auto-create-local-directory="true"
             local-filename-generator-expression="#this.toUpperCase() + '.a'"
+            scanner="myDirScanner"
             local-filter="myFilter"
             temporary-file-suffix=".writing"
             max-fetch-size="-1"
@@ -330,6 +331,7 @@ The inbound channel adapter first retrieves the file to a local directory and th
 Starting with _version 5.0_ you can now limit the number of files fetched from the FTP server when new file retrievals are needed.
 This can be beneficial when the target files are very large and/or when running in a clustered system with a persistent file list filter discussed below.
 Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved; see <<sftp-max-fetch>> for more information.
+Since _version 5.0_, you can also provide a custom `DirectoryScanner` implementation to the `inbound-channel-adapter` via the `scanner` attribute.
 
 Starting with _Spring Integration 3.0_, you can specify the `preserve-timestamp` attribute (default `false`); when `true`, the local file's modified timestamp will be set to the value retrieved from the server; otherwise it will be set to the current time.
 

src/reference/asciidoc/whats-new.adoc

@@ -168,8 +168,9 @@ See <<files>> for more information.
 
 ==== (S)FTP Changes
 
-The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.
+The Inbound Channel Adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.
 They also are configured with a `FileSystemPersistentAcceptOnceFileListFilter` in the `local-filter` by default.
+You can also provide a custom `DirectoryScanner` implementation to Inbound Channel Adapters via the newly introduced `scanner` attribute.
 
 The regex and pattern filters can now be configured to always pass directories.
 This can be useful when using recursion in the outbound gateways.