From 5a442162018f6afd5d0bc639c2696babf7c2e529 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 18 Oct 2016 10:32:23 -0400 Subject: [PATCH] INT-4140: File Streaming Adapter: Add maxFetchSize JIRA: https://jira.spring.io/browse/INT-4140 --- ...eStreamingInboundChannelAdapterParser.java | 1 + ...tractRemoteFileStreamingMessageSource.java | 29 +++++++++- .../file/remote/StreamingInboundTests.java | 56 +++++++++++++++++-- .../ftp/config/spring-integration-ftp-5.0.xsd | 2 +- ...boundChannelAdapterParserTests-context.xml | 1 + ...amingInboundChannelAdapterParserTests.java | 1 + .../config/spring-integration-sftp-5.0.xsd | 2 +- ...boundChannelAdapterParserTests-context.xml | 1 + ...amingInboundChannelAdapterParserTests.java | 1 + src/reference/asciidoc/ftp.adoc | 4 ++ src/reference/asciidoc/sftp.adoc | 4 ++ 11 files changed, 92 insertions(+), 10 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java index ca02be1eb7..6b6b338c1f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java @@ -62,6 +62,7 @@ public abstract class AbstractRemoteFileStreamingInboundChannelAdapterParser if (StringUtils.hasText(comparator)) { messageSourceBuilder.addConstructorArgReference(comparator); } + IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "max-fetch-size"); return messageSourceBuilder.getBeanDefinition(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java index de188b79d0..e02f259bbd 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -18,6 +18,7 @@ package org.springframework.integration.file.remote; import java.io.IOException; import java.io.InputStream; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -32,9 +33,10 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.IntegrationMessageHeaderAccessor; -import org.springframework.integration.endpoint.AbstractMessageSource; +import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.ReversibleFileListFilter; import org.springframework.integration.file.remote.session.Session; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -47,8 +49,8 @@ import org.springframework.util.Assert; * @since 4.3 * */ -public abstract class AbstractRemoteFileStreamingMessageSource extends AbstractMessageSource - implements BeanFactoryAware, InitializingBean { +public abstract class AbstractRemoteFileStreamingMessageSource + extends AbstractFetchLimitingMessageSource implements BeanFactoryAware, InitializingBean { private final RemoteFileTemplate remoteFileTemplate; @@ -147,6 +149,11 @@ public abstract class AbstractRemoteFileStreamingMessageSource extends Abstra return null; } + @Override + protected Object doReceive(int maxFetchSize) { + return doReceive(); + } + protected AbstractFileInfo poll() { if (this.toBeReceived.size() == 0) { listFiles(); @@ -164,7 +171,16 @@ public abstract class AbstractRemoteFileStreamingMessageSource extends Abstra private void listFiles() { String remoteDirectory = this.remoteDirectoryExpression.getValue(getEvaluationContext(), String.class); F[] files = this.remoteFileTemplate.list(remoteDirectory); + int maxFetchSize = getMaxFetchSize(); List filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files); + if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) { + rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); + List newList = new ArrayList<>(maxFetchSize); + for (int i = 0; i < maxFetchSize; i++) { + newList.add(filteredFiles.get(i)); + } + filteredFiles = newList; + } List> fileInfoList = asFileInfoList(filteredFiles); Iterator> iterator = fileInfoList.iterator(); while (iterator.hasNext()) { @@ -182,6 +198,13 @@ public abstract class AbstractRemoteFileStreamingMessageSource extends Abstra this.toBeReceived.addAll(fileInfoList); } + protected void rollbackFromFileToListEnd(List filteredFiles, F file) { + if (this.filter instanceof ReversibleFileListFilter) { + ((ReversibleFileListFilter) this.filter) + .rollback(file, filteredFiles); + } + } + abstract protected List> asFileInfoList(Collection files); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java index e1d4aea5d7..fb1878835b 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertNull; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import java.io.ByteArrayInputStream; @@ -36,6 +37,7 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.splitter.FileSplitter; @@ -54,7 +56,8 @@ public class StreamingInboundTests { @SuppressWarnings("unchecked") @Test public void testAllData() throws Exception { - Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null); + StringSessionFactory sessionFactory = new StringSessionFactory(); + Streamer streamer = new Streamer(new StringRemoteFileTemplate(sessionFactory), null); streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/foo"); streamer.afterPropertiesSet(); @@ -63,14 +66,47 @@ public class StreamingInboundTests { assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); - verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource()).close(); + // close after list, transform + verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(2)).close(); received = (Message) this.transformer.transform(streamer.receive()); assertEquals("baz\nqux", new String(received.getPayload())); assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); - verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource()).close(); + // close after transform + verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(3)).close(); + + verify(sessionFactory.getSession()).list("/foo"); + } + + @SuppressWarnings("unchecked") + @Test + public void testAllDataMaxFetch() throws Exception { + StringSessionFactory sessionFactory = new StringSessionFactory(); + Streamer streamer = new Streamer(new StringRemoteFileTemplate(sessionFactory), null); + streamer.setBeanFactory(mock(BeanFactory.class)); + streamer.setRemoteDirectory("/foo"); + streamer.setMaxFetchSize(1); + streamer.setFilter(new AcceptOnceFileListFilter<>()); + streamer.afterPropertiesSet(); + Message received = (Message) this.transformer.transform(streamer.receive()); + assertEquals("foo\nbar", new String(received.getPayload())); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + + // close after list, transform + verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(2)).close(); + + received = (Message) this.transformer.transform(streamer.receive()); + assertEquals("baz\nqux", new String(received.getPayload())); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + + // close after list, transform + verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(4)).close(); + + verify(sessionFactory.getSession(), times(2)).list("/foo"); } @SuppressWarnings("unchecked") @@ -97,7 +133,8 @@ public class StreamingInboundTests { assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); assertNull(out.receive(0)); - verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close(); + // close by list, splitter + verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(2)).close(); receivedStream = streamer.receive(); splitter.handleMessage(receivedStream); @@ -111,7 +148,8 @@ public class StreamingInboundTests { assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); assertNull(out.receive(0)); - verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close(); + // close by splitter + verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(3)).close(); } public static class Streamer extends AbstractRemoteFileStreamingMessageSource { @@ -191,9 +229,14 @@ public class StreamingInboundTests { public static class StringSessionFactory implements SessionFactory { + private Session session; + @SuppressWarnings("unchecked") @Override public Session getSession() { + if (this.session != null) { + return this.session; + } try { Session session = mock(Session.class); willReturn(new String[] { "/foo/foo", "/foo/bar" }).given(session).list("/foo"); @@ -209,6 +252,9 @@ public class StreamingInboundTests { willReturn(bar2).given(session).readRaw("/bar/bar"); given(session.finalizeRaw()).willReturn(true); + + this.session = session; + return session; } catch (Exception e) { 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 0cb7dd0e04..ca817d438f 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 @@ -164,7 +164,6 @@ - @@ -546,6 +545,7 @@ + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml index db90c2f0a2..83d137fe3d 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml @@ -22,6 +22,7 @@ filename-pattern="*.txt" remote-file-separator="X" comparator="comparator" + max-fetch-size="31" remote-directory-expression="'foo/bar'"> diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java index 7040442bd7..5866ff43eb 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java @@ -73,6 +73,7 @@ public class FtpStreamingInboundChannelAdapterParserTests { assertThat(TestUtils.getPropertyValue(source, "remoteFileSeparator", String.class), equalTo("X")); assertThat(TestUtils.getPropertyValue(source, "filter"), instanceOf(FtpSimplePatternFileListFilter.class)); assertSame(this.csf, TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory")); + assertEquals(31, TestUtils.getPropertyValue(source, "maxFetchSize")); } public static class TestSessionFactoryBean implements FactoryBean { 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 adbdc15651..02b4254682 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 @@ -168,7 +168,6 @@ - @@ -547,6 +546,7 @@ + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml index 407723092c..91d10c657f 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml @@ -22,6 +22,7 @@ filename-pattern="*.txt" remote-file-separator="X" comparator="comparator" + max-fetch-size="31" remote-directory-expression="'foo/bar'"> diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java index bc992f342f..aedacf743a 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java @@ -73,6 +73,7 @@ public class SftpStreamingInboundChannelAdapterParserTests { assertThat(TestUtils.getPropertyValue(source, "remoteFileSeparator", String.class), equalTo("X")); assertThat(TestUtils.getPropertyValue(source, "filter"), instanceOf(SftpSimplePatternFileListFilter.class)); assertSame(this.csf, TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory")); + assertEquals(31, TestUtils.getPropertyValue(source, "maxFetchSize")); } public static class TestSessionFactoryBean implements FactoryBean { diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index a737522527..74ba53a1df 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -446,6 +446,7 @@ See <> and <> for more information about thes filter="filter" remote-file-separator="/" comparator="comparator" + max-fetch-size="1" remote-directory-expression="'foo/bar'"> @@ -459,6 +460,8 @@ file being processed again, you can configure an `FtpPersistentFileListFilter` i If you don't actually want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter. If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`. +Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment; + ==== Configuring with Java Configuration The following Spring Boot application provides an example of configuring the inbound adapter using Java configuration: @@ -480,6 +483,7 @@ public class FtpJavaApplication { messageSource.setRemoteDirectory("ftpSource/"); messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming")); + messageSource.setMaxFetchSize(1); return messageSource; } diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 8950e29e85..9b1a1e0436 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -555,6 +555,7 @@ See <> and <> for more information about thes filter="filter" remote-file-separator="/" comparator="comparator" + max-fetch-size="1" remote-directory-expression="'foo/bar'"> @@ -568,6 +569,8 @@ file being processed again, you can configure an `SftpPersistentFileListFilter` If you don't actually want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter. If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`. +Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment; + ==== Configuring with Java Configuration The following Spring Boot application provides an example of configuring the inbound adapter using Java configuration: @@ -589,6 +592,7 @@ public class SftpJavaApplication { messageSource.setRemoteDirectory("sftpSource/"); messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming")); + messageSource.setMaxFetchSize(1); return messageSource; }