diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java index d1e283f52c..b105dd02e7 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java +++ b/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"); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java index 6e128c3bec..54f9be54cc 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java +++ b/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 extends AbstractFetchLimitingMessageSource implements Lifecycle { @@ -88,6 +90,10 @@ public abstract class AbstractInboundFileSynchronizingMessageSource private volatile FileListFilter localFileListFilter; + /** + * Whether the {@link DirectoryScanner} was explicitly set. + */ + private volatile boolean scannerExplicitlySet = false; public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer synchronizer) { this(synchronizer, null); @@ -145,6 +151,17 @@ public abstract class AbstractInboundFileSynchronizingMessageSource } } + /** + * 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 new SimpleMetadataStore(), getComponentName()); } FileListFilter 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); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java index 2e69dedfde..d6fdc1c3dc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java +++ b/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 sync = createLimitingSynchronizer(count); - - AbstractInboundFileSynchronizingMessageSource source = - new AbstractInboundFileSynchronizingMessageSource(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"); + AbstractInboundFileSynchronizingMessageSource source = createSource(sync); 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 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 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 source = createSource(count); + source.setUseWatchService(true); + source.setScanner(new HeadDirectoryScanner(1)); + source.afterPropertiesSet(); + } + + private AbstractInboundFileSynchronizingMessageSource createSource(AtomicInteger count) { + return createSource(createLimitingSynchronizer(count)); + } + + private AbstractInboundFileSynchronizingMessageSource createSource( + AbstractInboundFileSynchronizer sync) { + AbstractInboundFileSynchronizingMessageSource source = + new AbstractInboundFileSynchronizingMessageSource(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 createLimitingSynchronizer(final AtomicInteger count) { SessionFactory sf = new StringSessionFactory(); AbstractInboundFileSynchronizer sync = new AbstractInboundFileSynchronizer(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 diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd index d27110246f..3d46efe242 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd @@ -163,6 +163,19 @@ + + + + + + + + + Reference to a custom DirectoryScanner implementation. + + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml index 6d9abd6cc4..7049b54587 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml @@ -17,6 +17,7 @@ + + + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index 926afb88a1..8b2498c60d 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/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 genMethod = new AtomicReference(); 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; } + } } diff --git a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd index 85bbb4dd91..ba5265367b 100644 --- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd @@ -167,6 +167,19 @@ + + + + + + + + + Reference to a custom DirectoryScanner implementation. + + + diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 47d109e187..079b09c160 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/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 <> 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. diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index c25d9373ae..9025575796 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/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 <> 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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 4a12edbdf2..8118f5408e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -168,8 +168,9 @@ See <> 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.