diff --git a/README.md b/README.md index 23584ea..d5ea093 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,82 @@ An XML variant may look like: ```` +###Streaming Inbound Channel Adapter + +This adapter produces message with payloads of type `InputStream`, allowing S3 objects to be fetched without writing to the local file system. +Since the session remains open, the consuming application is responsible for closing the session when the file has been consumed. +The session is provided in the closeableResource header (`IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE`). Standard framework components, such as the `FileSplitter` and `StreamTransformer` will automatically close the session. + +The following Spring Boot application provides an example of configuring the S3 inbound streaming adapter using Java configuration: + +````java +@SpringBootApplication +public class S3JavaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(S3JavaApplication.class) + .web(false) + .run(args); + } + + @Autowired + private AmazonS3 amazonS3; + + @Bean + @InboundChannelAdapter(value = "s3Channel", poller = @Poller(fixedDelay = "100")) + public MessageSource s3InboundStreamingMessageSource() { + S3StreamingMessageSource messageSource = new S3StreamingMessageSource(template()); + messageSource.setRemoteDirectory(S3_BUCKET); + messageSource.setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), + "streaming")); + return messageSource; + } + + @Bean + @Transformer(inputChannel = "s3Channel", outputChannel = "data") + public org.springframework.integration.transformer.Transformer transformer() { + return new StreamTransformer(); + } + + @Bean + public S3RemoteFileTemplate template() { + return new S3RemoteFileTemplate(new S3SessionFactory(amazonS3)); + } + + @Bean + public PollableChannel s3Channel() { + return new QueueChannel(); + } +} +```` + +An XML variant may look like: + +````xml + + + + + + + + + + + + +```` + +Only one of `filename-pattern`, `filename-regex` or `filter` is allowed. + +> NOTE: Unlike the non-streaming inbound channel adapter, this adapter does not prevent duplicates by default. +> If you do not delete the remote file and you wish to prevent the file being processed again, you can configure an `S3PersistentFileListFilter` in the `filter` attribute. +> 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`. + ###Outbound Channel Adapter The S3 Outbound Channel Adapter is represented by the `S3MessageHandler` (`` diff --git a/build.gradle b/build.gradle index 94d3324..e8b556a 100644 --- a/build.gradle +++ b/build.gradle @@ -73,12 +73,12 @@ ext.javadocLinks = [ ] as String[] jacoco { - toolVersion = "0.7.6.201602180812" + toolVersion = "0.7.7.201606060606" } checkstyle { configFile = file('src/checkstyle/checkstyle.xml') - toolVersion = '6.18' + toolVersion = "6.16.1" } dependencies { diff --git a/src/main/java/org/springframework/integration/aws/config/xml/AwsNamespaceHandler.java b/src/main/java/org/springframework/integration/aws/config/xml/AwsNamespaceHandler.java index 4a9b3f9..21fd641 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/AwsNamespaceHandler.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/AwsNamespaceHandler.java @@ -32,6 +32,7 @@ public class AwsNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("s3-outbound-channel-adapter", new S3OutboundChannelAdapterParser()); registerBeanDefinitionParser("s3-outbound-gateway", new S3OutboundGatewayParser()); registerBeanDefinitionParser("s3-inbound-channel-adapter", new S3InboundChannelAdapterParser()); + registerBeanDefinitionParser("s3-inbound-streaming-channel-adapter", new S3InboundStreamingChannelAdapterParser()); registerBeanDefinitionParser("sqs-outbound-channel-adapter", new SqsOutboundChannelAdapterParser()); registerBeanDefinitionParser("sqs-message-driven-channel-adapter", new SqsMessageDrivenChannelAdapterParser()); registerBeanDefinitionParser("sns-inbound-channel-adapter", new SnsInboundChannelAdapterParser()); diff --git a/src/main/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParser.java b/src/main/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParser.java new file mode 100644 index 0000000..24f7ce7 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParser.java @@ -0,0 +1,56 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.config.xml; + +import org.springframework.integration.aws.inbound.S3InboundStreamingMessageSource; +import org.springframework.integration.aws.support.S3RemoteFileTemplate; +import org.springframework.integration.aws.support.filters.S3RegexPatternFileListFilter; +import org.springframework.integration.aws.support.filters.S3SimplePatternFileListFilter; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.config.AbstractRemoteFileStreamingInboundChannelAdapterParser; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.RemoteFileOperations; + +/** + * Parser for the AWS 's3-inbound-streaming-channel-adapter' element. + * + * @author Christian Tzolov + * @since 1.1 + */ +public class S3InboundStreamingChannelAdapterParser extends AbstractRemoteFileStreamingInboundChannelAdapterParser { + + @Override + protected Class> getTemplateClass() { + return S3RemoteFileTemplate.class; + } + + @Override + protected Class> getMessageSourceClass() { + return S3InboundStreamingMessageSource.class; + } + + @Override + protected Class> getSimplePatternFileListFilterClass() { + return S3SimplePatternFileListFilter.class; + } + + @Override + protected Class> getRegexPatternFileListFilterClass() { + return S3RegexPatternFileListFilter.class; + } + +} diff --git a/src/main/java/org/springframework/integration/aws/inbound/S3InboundStreamingMessageSource.java b/src/main/java/org/springframework/integration/aws/inbound/S3InboundStreamingMessageSource.java new file mode 100644 index 0000000..6091805 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/inbound/S3InboundStreamingMessageSource.java @@ -0,0 +1,63 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.inbound; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; + +import org.springframework.integration.aws.support.S3FileInfo; +import org.springframework.integration.file.remote.AbstractFileInfo; +import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource; +import org.springframework.integration.file.remote.RemoteFileTemplate; + +import com.amazonaws.services.s3.model.S3ObjectSummary; + +/** + * A {@link AbstractRemoteFileStreamingMessageSource} implementation for the Amazon S3. + * + * @author Christian Tzolov + * @since 1.1 + */ +public class S3InboundStreamingMessageSource extends AbstractRemoteFileStreamingMessageSource { + + public S3InboundStreamingMessageSource(RemoteFileTemplate template) { + super(template, null); + } + + public S3InboundStreamingMessageSource(RemoteFileTemplate template, + Comparator> comparator) { + super(template, comparator); + } + + @Override + protected List> asFileInfoList(Collection collection) { + + List> canonicalFiles = new ArrayList>(); + for (S3ObjectSummary s3ObjectSummary : collection) { + canonicalFiles.add(new S3FileInfo(s3ObjectSummary)); + } + return canonicalFiles; + } + + @Override + public String getComponentType() { + return "aws:s3-inbound-streaming-channel-adapter"; + } + +} diff --git a/src/main/java/org/springframework/integration/aws/support/S3FileInfo.java b/src/main/java/org/springframework/integration/aws/support/S3FileInfo.java new file mode 100644 index 0000000..ab4c234 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/support/S3FileInfo.java @@ -0,0 +1,90 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.support; + +import java.util.Date; + +import org.springframework.integration.file.remote.AbstractFileInfo; +import org.springframework.util.Assert; + +import com.amazonaws.services.s3.model.S3ObjectSummary; + +/** + * An Amazon S3 {@link org.springframework.integration.file.remote.FileInfo} implementation. + * @author Christian Tzolov + * @since 1.1 + */ +public class S3FileInfo extends AbstractFileInfo { + + private final S3ObjectSummary s3ObjectSummary; + + public S3FileInfo(S3ObjectSummary s3ObjectSummary) { + Assert.notNull(s3ObjectSummary, "s3ObjectSummary must not be null"); + this.s3ObjectSummary = s3ObjectSummary; + } + + @Override + public boolean isDirectory() { + return false; + } + + @Override + public boolean isLink() { + return false; + } + + @Override + public long getSize() { + return this.s3ObjectSummary.getSize(); + } + + @Override + public long getModified() { + return this.s3ObjectSummary.getLastModified().getTime(); + } + + @Override + public String getFilename() { + return this.s3ObjectSummary.getKey(); + } + + /** + * A permissions representation string. + * Throws {@link UnsupportedOperationException} to avoid extra + * {@link com.amazonaws.services.s3.AmazonS3#getObjectAcl} REST call. + * The target application amy choose to do that by its logic. + * @return the permissions representation string. + */ + @Override + public String getPermissions() { + throw new UnsupportedOperationException("Use [AmazonS3.getObjectAcl()] to obtain permissions."); + } + + @Override + public S3ObjectSummary getFileInfo() { + return this.s3ObjectSummary; + } + + @Override + public String toString() { + return "FileInfo [isDirectory=" + isDirectory() + ", isLink=" + isLink() + + ", Size=" + getSize() + ", ModifiedTime=" + + new Date(getModified()) + ", Filename=" + getFilename() + + ", RemoteDirectory=" + getRemoteDirectory() + "]"; + } + +} diff --git a/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.1.xsd b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.1.xsd index a3bfdf6..f293490 100644 --- a/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.1.xsd +++ b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.1.xsd @@ -433,6 +433,144 @@ + + + + Configures a 'SourcePollingChannelAdapter' Endpoint for the + 'org.springframework.integration.aws.inbound.S3InboundStreamingMessageSource'. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Identifies channel attached to this adapter. + The channel to which messages will be sent + by this adapter. + + + + + + + Allows you to provide a file name pattern to + determine the file names + that need to be scanned. + This is based on + simple pattern matching (e.g., "*.txt, fo*.txt" + etc.) + + + + + + + Allows you to provide a Regular Expression to + determine the file names + that need to be scanned. + (e.g., + "f[o]+\.txt" etc.) + + + + + + + + + + + + Allows you to specify a reference to a + [org.springframework.integration.file.filters.FileListFilter] + bean. This filter is applied to files on the remote server and + only files that pass the filter are retrieved. + + + + + + + + + + + + + + + + + + + + + + + Allows you to provide remote file/directory + separator character. DEFAULT: + '/' + + + + + + + Identifies the remote directory path (e.g., "/remote/mytransfers") + Mutually exclusive with 'remote-directory-expression'. + + + + + + + Specify a SpEL expression which + will be used to evaluate the directory + path to where the files will be transferred + (e.g., "headers.['remote_dir'] + '/myTransfers'" for outbound endpoints) + There is no root object (message) for inbound endpoints + (e.g., "@someBean.fetchDirectory"); + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParserTests-context.xml new file mode 100644 index 0000000..08a0beb --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParserTests-context.xml @@ -0,0 +1,43 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParserTests.java new file mode 100644 index 0000000..e2d5c16 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParserTests.java @@ -0,0 +1,103 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.config.xml; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.lang.reflect.Method; +import java.util.Comparator; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.expression.Expression; +import org.springframework.integration.aws.inbound.S3InboundStreamingMessageSource; +import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.ReflectionUtils; + +/** + * @author Christian Tzolov + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class S3InboundStreamingChannelAdapterParserTests { + + @Autowired + private SourcePollingChannelAdapter s3Inbound; + + @Autowired + private Comparator comparator; + + @Autowired + private MessageChannel s3Channel; + + @Autowired + private S3PersistentAcceptOnceFileListFilter acceptOnceFilter; + + @Autowired + private SessionFactory s3SessionFactory; + + @Test + public void testS3StreamingInboundChannelAdapterComplete() throws Exception { + + assertThat(TestUtils.getPropertyValue(this.s3Inbound, "autoStartup", Boolean.class)).isFalse(); + assertThat(this.s3Inbound.getComponentName()).isEqualTo("s3Inbound"); + assertThat(this.s3Inbound.getComponentType()).isEqualTo("aws:s3-inbound-streaming-channel-adapter"); + assertThat(TestUtils.getPropertyValue(this.s3Inbound, "outputChannel")).isSameAs(this.s3Channel); + + S3InboundStreamingMessageSource source = TestUtils.getPropertyValue(this.s3Inbound, "source", + S3InboundStreamingMessageSource.class); + + assertThat(TestUtils.getPropertyValue(source, "remoteDirectoryExpression", Expression.class) + .getExpressionString()) + .isEqualTo("foo/bar"); + + assertThat(TestUtils.getPropertyValue(source, "comparator")).isSameAs(this.comparator); + String remoteFileSeparator = (String) TestUtils.getPropertyValue(source, "remoteFileSeparator"); + assertThat(remoteFileSeparator).isNotNull(); + assertThat(remoteFileSeparator).isEqualTo("/"); + + S3PersistentAcceptOnceFileListFilter filter = TestUtils.getPropertyValue(source, "filter", + S3PersistentAcceptOnceFileListFilter.class); + assertThat(filter).isSameAs(this.acceptOnceFilter); + assertThat(TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory")) + .isSameAs(this.s3SessionFactory); + + final AtomicReference genMethod = new AtomicReference(); + ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class, new ReflectionUtils.MethodCallback() { + + @Override + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + if ("generateLocalFileName".equals(method.getName())) { + method.setAccessible(true); + genMethod.set(method); + } + } + + }); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/inbound/S3InboundStreamingChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/inbound/S3InboundStreamingChannelAdapterTests.java new file mode 100644 index 0000000..047ee73 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/inbound/S3InboundStreamingChannelAdapterTests.java @@ -0,0 +1,195 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.inbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Matchers.any; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; + +import org.apache.commons.io.IOUtils; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.aws.support.S3RemoteFileTemplate; +import org.springframework.integration.aws.support.S3SessionFactory; +import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.FileCopyUtils; + +import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.model.ListObjectsRequest; +import com.amazonaws.services.s3.model.ObjectListing; +import com.amazonaws.services.s3.model.S3Object; +import com.amazonaws.services.s3.model.S3ObjectSummary; + +/** + * @author Christian Tzolov + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +@DirtiesContext +public class S3InboundStreamingChannelAdapterTests { + + @ClassRule + public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + + private static final String S3_BUCKET = "S3_BUCKET"; + + private static List S3_OBJECTS; + + @Autowired + private PollableChannel s3FilesChannel; + + @BeforeClass + public static void setup() throws IOException { + File remoteFolder = TEMPORARY_FOLDER.newFolder("remote"); + + File aFile = new File(remoteFolder, "a.test"); + FileCopyUtils.copy("Hello".getBytes(), aFile); + File bFile = new File(remoteFolder, "b.test"); + FileCopyUtils.copy("Bye".getBytes(), bFile); + + S3_OBJECTS = new ArrayList<>(); + + for (File file : remoteFolder.listFiles()) { + S3Object s3Object = new S3Object(); + s3Object.setBucketName(S3_BUCKET); + s3Object.setKey("subdir/" + file.getName()); + s3Object.setObjectContent(new FileInputStream(file)); + + S3_OBJECTS.add(s3Object); + } + } + + @Test + public void testS3InboundChannelAdapter() throws IOException { + Message message = this.s3FilesChannel.receive(10000); + assertThat(message).isNotNull(); + assertThat(message.getPayload()).isInstanceOf(InputStream.class); + assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("subdir/a.test"); + + InputStream inputStreamA = (InputStream) message.getPayload(); + assertThat(inputStreamA).isNotNull(); + assertThat(IOUtils.toString(inputStreamA)).isEqualTo("Hello"); + + message = this.s3FilesChannel.receive(10000); + assertThat(message).isNotNull(); + assertThat(message.getPayload()).isInstanceOf(InputStream.class); + assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("subdir/b.test"); + InputStream inputStreamB = (InputStream) message.getPayload(); + assertThat(IOUtils.toString(inputStreamB)).isEqualTo("Bye"); + + assertThat(this.s3FilesChannel.receive(10000)).isNull(); + } + + @Configuration + @EnableIntegration + public static class Config { + + @Bean + public AmazonS3 amazonS3() { + AmazonS3 amazonS3 = Mockito.mock(AmazonS3.class); + + willAnswer(new Answer() { + + @Override + public ObjectListing answer(InvocationOnMock invocation) throws Throwable { + ObjectListing objectListing = new ObjectListing(); + List objectSummaries = objectListing.getObjectSummaries(); + for (S3Object s3Object : S3_OBJECTS) { + S3ObjectSummary s3ObjectSummary = new S3ObjectSummary(); + s3ObjectSummary.setBucketName(S3_BUCKET); + s3ObjectSummary.setKey(s3Object.getKey()); + + s3ObjectSummary.setLastModified(new Date(new File(s3Object.getKey()).lastModified())); + objectSummaries.add(s3ObjectSummary); + } + return objectListing; + } + + }).given(amazonS3).listObjects(any(ListObjectsRequest.class)); + + for (final S3Object s3Object : S3_OBJECTS) { + willAnswer(new Answer() { + + @Override + public S3Object answer(InvocationOnMock invocation) throws Throwable { + return s3Object; + } + + }).given(amazonS3).getObject(S3_BUCKET, s3Object.getKey()); + } + + return amazonS3; + } + + + @Bean + @InboundChannelAdapter(value = "s3FilesChannel", poller = @Poller(fixedDelay = "100")) + public S3InboundStreamingMessageSource s3InboundStreamingMessageSource(AmazonS3 amazonS3) { + + S3SessionFactory s3SessionFactory = new S3SessionFactory(amazonS3); + + S3RemoteFileTemplate s3FileTemplate = new S3RemoteFileTemplate(s3SessionFactory); + + s3FileTemplate.setUseTemporaryFileName(false); + + S3InboundStreamingMessageSource s3MessageSource = new S3InboundStreamingMessageSource(s3FileTemplate); + + s3MessageSource.setRemoteDirectory(S3_BUCKET); + + s3MessageSource.setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), + "streaming")); + + return s3MessageSource; + } + + @Bean + public PollableChannel s3FilesChannel() { + return new QueueChannel(); + } + + } + +}