Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4140: File Streaming Adapter: Add maxFetchSize

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4140
This commit is contained in:
Gary Russell
2016-10-18 10:32:23 -04:00
parent 3c284871c3
commit 5a44216201
11 changed files with 92 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java
View File

@@ -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();
}

29
spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
View File

@@ -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<F> extends AbstractMessageSource<InputStream>
implements BeanFactoryAware, InitializingBean {
public abstract class AbstractRemoteFileStreamingMessageSource<F>
extends AbstractFetchLimitingMessageSource<InputStream> implements BeanFactoryAware, InitializingBean {
private final RemoteFileTemplate<F> remoteFileTemplate;
@@ -147,6 +149,11 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F> extends Abstra
return null;
}
@Override
protected Object doReceive(int maxFetchSize) {
return doReceive();
}
protected AbstractFileInfo<F> poll() {
if (this.toBeReceived.size() == 0) {
listFiles();
@@ -164,7 +171,16 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F> extends Abstra
private void listFiles() {
String remoteDirectory = this.remoteDirectoryExpression.getValue(getEvaluationContext(), String.class);
F[] files = this.remoteFileTemplate.list(remoteDirectory);
int maxFetchSize = getMaxFetchSize();
List<F> filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files);
if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) {
rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
List<F> newList = new ArrayList<>(maxFetchSize);
for (int i = 0; i < maxFetchSize; i++) {
newList.add(filteredFiles.get(i));
}
filteredFiles = newList;
}
List<AbstractFileInfo<F>> fileInfoList = asFileInfoList(filteredFiles);
Iterator<AbstractFileInfo<F>> iterator = fileInfoList.iterator();
while (iterator.hasNext()) {
@@ -182,6 +198,13 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F> extends Abstra
this.toBeReceived.addAll(fileInfoList);
}
protected void rollbackFromFileToListEnd(List<F> filteredFiles, F file) {
if (this.filter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) this.filter)
.rollback(file, filteredFiles);
}
}
abstract protected List<AbstractFileInfo<F>> asFileInfoList(Collection<F> files);
}

56
spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
View File

@@ -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<byte[]>) 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<byte[]> received = (Message<byte[]>) 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<byte[]>) 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<String> {
@@ -191,9 +229,14 @@ public class StreamingInboundTests {
public static class StringSessionFactory implements SessionFactory<String> {
private Session<String> session;
@SuppressWarnings("unchecked")
@Override
public Session<String> getSession() {
if (this.session != null) {
return this.session;
}
try {
Session<String> 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) {

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

@@ -164,7 +164,6 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="tempSuffixGroup" />
<xsd:attributeGroup ref="integration:maxFetchGroup" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -546,6 +545,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:maxFetchGroup" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml
View File

@@ -22,6 +22,7 @@
filename-pattern="*.txt"
remote-file-separator="X"
comparator="comparator"
max-fetch-size="31"
remote-directory-expression="'foo/bar'">
<int:poller fixed-rate="1000" />
</int-ftp:inbound-streaming-channel-adapter>

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java
View File

@@ -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<DefaultFtpSessionFactory> {

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

@@ -168,7 +168,6 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="tempSuffixGroup" />
<xsd:attributeGroup ref="integration:maxFetchGroup" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -547,6 +546,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:maxFetchGroup" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

1
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml
View File

@@ -22,6 +22,7 @@
filename-pattern="*.txt"
remote-file-separator="X"
comparator="comparator"
max-fetch-size="31"
remote-directory-expression="'foo/bar'">
<int:poller fixed-rate="1000" />
</int-sftp:inbound-streaming-channel-adapter>

1
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java
View File

@@ -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<DefaultSftpSessionFactory> {

4
src/reference/asciidoc/ftp.adoc
View File

@@ -446,6 +446,7 @@ See <<file-splitter>> and <<stream-transformer>> for more information about thes
filter="filter"
remote-file-separator="/"
comparator="comparator"
max-fetch-size="1"
remote-directory-expression="'foo/bar'">
<int:poller fixed-rate="1000" />
</int-ftp:inbound-streaming-channel-adapter>
@@ -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;
}

4
src/reference/asciidoc/sftp.adoc
View File

@@ -555,6 +555,7 @@ See <<file-splitter>> and <<stream-transformer>> for more information about thes
filter="filter"
remote-file-separator="/"
comparator="comparator"
max-fetch-size="1"
remote-directory-expression="'foo/bar'">
<int:poller fixed-rate="1000" />
</int-sftp:inbound-streaming-channel-adapter>
@@ -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;
}