Issue #30 : Add S3 inbound streaming channel adapter.

Fixes GH-30 (https://github.com/spring-projects/spring-integration-aws/issues/30)

* Add S3InboundStreamingMessageSource and related S3FileInfo
* Extend spring-integration-aws-1.1.xsd with s3-inbound-streaming-channel-adapter tag.
* Add S3StreamingInboundChannelAdapterParser and s3-inbound-streaming-channel-adapter handler.
* Add S3 streaming spring tests.
* Add S3 streaming documentation to README.md.

Issue #30: Resolve code style and formatting issues

Polishing:
* Revert some `build.gradle` changes
* Add JavaDoc to the `S3FileInfo.getPermissions()`

README.md

@@ -115,6 +115,82 @@ An XML variant may look like:
 </int-aws:s3-inbound-channel-adapter>
 ````
 
+###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<InputStream> 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
+<bean id="metadataStore" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
+
+<bean id="acceptOnceFilter" class="org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter">
+	<constructor-arg index="0" ref="metadataStore"/>
+	<constructor-arg index="1" value="streaming"/>
+</bean>
+
+<bean id="s3SessionFactory" class="org.springframework.integration.aws.support.S3SessionFactory"/>
+
+<int-aws:s3-inbound-streaming-channel-adapter channel="s3Channel"
+                   session-factory="s3SessionFactory"
+                   filter="acceptOnceFilter"
+                   remote-directory-expression="'my_bucket'">
+    <int:poller fixed-rate="1000"/>
+</int-aws:s3-inbound-streaming-channel-adapter>
+````
+
+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` (`<int-aws:s3-outbound-channel-adapter>`

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 {

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

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<? extends RemoteFileOperations<?>> getTemplateClass() {
+		return S3RemoteFileTemplate.class;
+	}
+
+	@Override
+	protected Class<? extends MessageSource<?>> getMessageSourceClass() {
+		return S3InboundStreamingMessageSource.class;
+	}
+
+	@Override
+	protected Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass() {
+		return S3SimplePatternFileListFilter.class;
+	}
+
+	@Override
+	protected Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass() {
+		return S3RegexPatternFileListFilter.class;
+	}
+
+}

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<S3ObjectSummary> {
+
+	public S3InboundStreamingMessageSource(RemoteFileTemplate<S3ObjectSummary> template) {
+		super(template, null);
+	}
+
+	public S3InboundStreamingMessageSource(RemoteFileTemplate<S3ObjectSummary> template,
+			Comparator<AbstractFileInfo<S3ObjectSummary>> comparator) {
+		super(template, comparator);
+	}
+
+	@Override
+	protected List<AbstractFileInfo<S3ObjectSummary>> asFileInfoList(Collection<S3ObjectSummary> collection) {
+
+		List<AbstractFileInfo<S3ObjectSummary>> canonicalFiles = new ArrayList<AbstractFileInfo<S3ObjectSummary>>();
+		for (S3ObjectSummary s3ObjectSummary : collection) {
+			canonicalFiles.add(new S3FileInfo(s3ObjectSummary));
+		}
+		return canonicalFiles;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "aws:s3-inbound-streaming-channel-adapter";
+	}
+
+}

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<S3ObjectSummary> {
+
+	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() + "]";
+	}
+
+}

src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.1.xsd

@@ -433,6 +433,144 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="s3-inbound-streaming-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures a 'SourcePollingChannelAdapter' Endpoint for the
+				'org.springframework.integration.aws.inbound.S3InboundStreamingMessageSource'.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="base-inbound-adapter-type">
+					<xsd:attribute name="comparator" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								Specify a Comparator to be used when ordering Files. If none is provided, the
+								order in which files are processed is the order they are received from the
+								S3 server. The generic type of the Comparator must be 'S3FileInfo'.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:complexType name="base-inbound-adapter-type">
+		<xsd:complexContent>
+			<xsd:extension base="base-adapter-type">
+				<xsd:sequence>
+					<xsd:element ref="integration:poller" minOccurs="0"
+								 maxOccurs="1" />
+				</xsd:sequence>
+				<xsd:attribute name="channel" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
+							</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+							Identifies channel attached to this adapter.
+							The channel to which messages will be sent
+							by this adapter.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="filename-pattern" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							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.)
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="filename-regex" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							Allows you to provide a Regular Expression to
+							determine the file names
+							that need to be scanned.
+							(e.g.,
+							"f[o]+\.txt" etc.)
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="filter" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type
+										type="org.springframework.integration.file.filters.FileListFilter" />
+							</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+							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.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+			</xsd:extension>
+		</xsd:complexContent>
+	</xsd:complexType>
+
+	<xsd:complexType name="base-adapter-type">
+		<xsd:attribute name="id" type="xsd:string" />
+		<xsd:attribute name="session-factory" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type
+								type="org.springframework.integration.file.remote.session.SessionFactory" />
+					</tool:annotation>
+				</xsd:appinfo>
+				<xsd:documentation><![CDATA[
+					Reference to an [org.springframework.integration.file.remote.session.SessionFactory] bean with
+					a [com.amazonaws.services.s3.model.S3ObjectSummary] generic type parameter.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="remote-file-separator" type="xsd:string"
+					   default="/">
+			<xsd:annotation>
+				<xsd:documentation>
+					Allows you to provide remote file/directory
+					separator character. DEFAULT:
+					'/'
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="remote-directory" type="xsd:string" use="optional">
+			<xsd:annotation>
+				<xsd:documentation>
+					Identifies the remote directory path (e.g., "/remote/mytransfers")
+					Mutually exclusive with 'remote-directory-expression'.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="remote-directory-expression"
+					   type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					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");
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
+	</xsd:complexType>
+
 	<xsd:element name="sqs-outbound-channel-adapter">
 		<xsd:complexType>
 			<xsd:annotation>

src/test/java/org/springframework/integration/aws/config/xml/S3InboundStreamingChannelAdapterParserTests-context.xml

@@ -0,0 +1,43 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	   http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd
+	   http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<bean id="s3" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="com.amazonaws.services.s3.AmazonS3"/>
+	</bean>
+
+	<bean id="s3SessionFactory" class="org.springframework.integration.aws.support.S3SessionFactory">
+		<constructor-arg ref="s3"/>
+	</bean>
+
+	<bean id="comparator" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="java.util.Comparator"/>
+	</bean>
+
+	<int:channel id="s3Channel">
+		<int:queue/>
+	</int:channel>
+
+	<bean id="metadataStore" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
+
+	<bean id="acceptOnceFilter" class="org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter">
+		<constructor-arg index="0" ref="metadataStore"/>
+		<constructor-arg index="1" value="streaming"/>
+	</bean>
+
+	<int-aws:s3-inbound-streaming-channel-adapter id="s3Inbound"
+												  session-factory="s3SessionFactory"
+												  channel="s3Channel"
+												  auto-startup="false"
+												  comparator="comparator"
+												  filter="acceptOnceFilter"
+												  remote-directory-expression="foo/bar">
+		<int:poller fixed-rate="1000"/>
+	</int-aws:s3-inbound-streaming-channel-adapter>
+
+</beans>

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<Method> genMethod = new AtomicReference<Method>();
+		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);
+				}
+			}
+
+		});
+	}
+
+}

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<S3Object> 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<ObjectListing>() {
+
+				@Override
+				public ObjectListing answer(InvocationOnMock invocation) throws Throwable {
+					ObjectListing objectListing = new ObjectListing();
+					List<S3ObjectSummary> 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<S3Object>() {
+
+					@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();
+		}
+
+	}
+
+}