INT-4142: Backport Streaming (S)FTP

JIRA: https://jira.spring.io/browse/INT-4142 INT-4015: Streaming Remote File Inbound Adapter Initial commit. Reworked to emit an input stream and use the file splitter. Add StreamTransformer. Add CLOSABLE_RESOURCE header so we can close the session automatically. Implement INT-3854, FTP, SFTP (S)FTP Namespace Changes Docs - also fixes a PDF overflow Polishing - PR Comments checkstyle fixes Polishing - Add Namespace for StreamParser Polishing - PR Comments Fix Streaming (S)FTP Tests https://build.spring.io/browse/INT-B43-190/ Add `AcceptOneFileListFilter`s. Polishing polishing Polishing * Polish `.travis.yml` to avoid unnecessary work on Travis CI
2016-04-27 15:10:02 -04:00
parent 3bd15cf7c0
commit 5b80ff499c
39 changed files with 1751 additions and 313 deletions
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,7 +5,14 @@ services:
  - mongodb
  - rabbitmq
  - redis-server
+before_cache:
+  - rm -f $HOME/.gradle/caches/modules-2/modules-2.lock
+cache:
+  directories:
+    - $HOME/.gradle/caches/
+    - $HOME/.gradle/wrapper/
+install: true
 env:
-  - TERM=dumb SI_FATAL_WHEN_NO_BEANFACTORY=true
-#script:
-#  - ./gradlew build --parallel
+  - TERM=dumb SI_FATAL_WHEN_NO_BEANFACTORY=true NO_REFERENCE_TASK=true
+script:
+    - ./gradlew check --refresh-dependencies --no-daemon
--- a/build.gradle
+++ b/build.gradle
@@ -355,6 +355,7 @@ project('spring-integration-ftp') {
 		compile "org.springframework:spring-context-support:$springVersion"
 		compile("javax.activation:activation:$javaxActivationVersion", optional)
 		testCompile "org.apache.ftpserver:ftpserver-core:$ftpServerVersion"
+		testCompile project(":spring-integration-file").sourceSets.test.output
 	}
 }

@@ -574,6 +575,7 @@ project('spring-integration-sftp') {
 		compile "org.springframework:spring-context-support:$springVersion"
 		compile("javax.activation:activation:$javaxActivationVersion", optional)
 		testCompile "org.apache.sshd:sshd-core:$apacheSshdVersion"
+		testCompile project(":spring-integration-file").sourceSets.test.output
 	}
 }

--- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
@@ -53,6 +53,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
 		registerBeanDefinitionParser("json-to-object-transformer", new JsonToObjectTransformerParser());
 		registerBeanDefinitionParser("payload-serializing-transformer", new PayloadSerializingTransformerParser());
 		registerBeanDefinitionParser("payload-deserializing-transformer", new PayloadDeserializingTransformerParser());
+		registerBeanDefinitionParser("stream-transformer", new StreamTransformerParser());
 		registerBeanDefinitionParser("claim-check-in", new ClaimCheckInParser());
 		registerBeanDefinitionParser("syslog-to-map-transformer", new SyslogToMapTransformerParser());
 		registerBeanDefinitionParser("claim-check-out", new ClaimCheckOutParser());
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StreamTransformerParser.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StreamTransformerParser.java
@@ -0,0 +1,35 @@
+/*
+ * 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.config.xml;
+
+import org.springframework.integration.transformer.StreamTransformer;
+
+/**
+ * Parser for {@code <stream-transformer/>} element.
+ *
+ * @author Gary Russell
+ * @since 4.2.12
+ *
+ */
+public class StreamTransformerParser extends ObjectToStringTransformerParser {
+
+	@Override
+	protected String getTransformerClassName() {
+		return StreamTransformer.class.getName();
+	}
+
+}
--- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd
+++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd
@@ -1816,6 +1816,8 @@
 					<xsd:element name="map-to-object-transformer" type="map-to-object-transformer-type"/>
 					<xsd:element name="object-to-json-transformer" type="object-to-json-transformer-type"/>
 					<xsd:element name="json-to-object-transformer" type="json-to-object-transformer-type"/>
+					<xsd:element name="stream-transformer" type="stream-transformer-type"/>
+					<xsd:element name="syslog-to-map-transformer" type="specialized-transformer-type"/>
 					<xsd:element name="claim-check-in" type="claimCheckInTypeChain"/>
 					<xsd:element name="claim-check-out" type="claimCheckOutTypeChain"/>
 					<xsd:element name="control-bus" type="control-bus-type"/>
@@ -2661,11 +2663,49 @@
 			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
-			<xsd:attributeGroup ref="inputOutputChannelGroup" />
-			<xsd:attribute name="id" type="xsd:string" />
+			<xsd:complexContent>
+				<xsd:extension base="specialized-transformer-type">
+					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+				</xsd:extension>
+			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>

+	<xsd:element name="stream-transformer">
+		<xsd:annotation>
+			<xsd:documentation>
+				Defines a Consumer Endpoint for the
+				'org.springframework.integration.transformer.StreamTransformer'
+				that converts an 'InputStream' payload to a byte[] or String.
+				Providing a 'charset' signals that the conversion to String is
+				required.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="stream-transformer-type">
+					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:complexType name="stream-transformer-type">
+		<xsd:complexContent>
+			<xsd:extension base="specialized-transformer-type">
+				<xsd:attribute name="charset" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							Allows you to specify the Charset (e.g., US-ASCII,
+							ISO-8859-1, UTF-8) to be used when transforming byte[].
+							None by default, meaning the payload will be byte[].
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+			</xsd:extension>
+		</xsd:complexContent>
+	</xsd:complexType>
+
    <!-- Claim Check -->

 	<xsd:element name="claim-check-in" type="claimCheckInType">
@@ -2762,8 +2802,7 @@
 	</xsd:complexType>

 	<xsd:complexType name="specialized-transformer-type">
-		<xsd:choice minOccurs="1" maxOccurs="unbounded">
-			<xsd:any processContents="strict" namespace="##other" minOccurs="0" maxOccurs="unbounded" />
+		<xsd:choice minOccurs="0" maxOccurs="1">
 			<xsd:element ref="poller" />
 		</xsd:choice>
 		<xsd:attribute name="id" type="xsd:string" />
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToStringTransformerParserTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToStringTransformerParserTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-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.
@@ -21,22 +21,26 @@ import static org.junit.Assert.assertNotNull;

 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
-import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.messaging.support.GenericMessage;
-import org.springframework.integration.test.util.TestUtils;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

 /**
 * @author Mark Fisher
+ * @author Gary Russell
 */
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class ObjectToStringTransformerParserTests {

 	@Autowired
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml
@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<channel id="directInput"/>
+
+	<channel id="queueInput">
+		<queue capacity="1"/>
+	</channel>
+
+	<channel id="output">
+		<queue capacity="1"/>
+	</channel>
+
+	<stream-transformer input-channel="directInput" output-channel="output"/>
+
+	<stream-transformer input-channel="queueInput" output-channel="output">
+		<poller fixed-delay="10000"/>
+	</stream-transformer>
+
+	<chain input-channel="charsetChannel" output-channel="output">
+		<stream-transformer id="withCharset" charset="UTF-8" />
+	</chain>
+
+</beans:beans>
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java
@@ -0,0 +1,88 @@
+/*
+ * 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.config.xml;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Mark Fisher
+ * @author Gary Russell
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class StreamTransformerParserTests {
+
+	@Autowired
+	@Qualifier("directInput")
+	private MessageChannel directInput;
+
+	@Autowired
+	@Qualifier("charsetChannel")
+	private MessageChannel charsetChannel;
+
+	@Autowired
+	@Qualifier("queueInput")
+	private MessageChannel queueInput;
+
+	@Autowired
+	@Qualifier("output")
+	private PollableChannel output;
+
+	@Test
+	public void directChannelWithStringMessage() {
+		this.directInput.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
+		Message<?> result = output.receive(0);
+		assertNotNull(result);
+		assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload());
+	}
+
+	@Test
+	public void queueChannelWithStringMessage() {
+		this.queueInput.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
+		Message<?> result = output.receive(3000);
+		assertNotNull(result);
+		assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload());
+	}
+
+	@Test
+	public void charset() {
+		this.charsetChannel.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
+		Message<?> result = output.receive(0);
+		assertNotNull(result);
+		assertEquals("foo", result.getPayload());
+	}
+
+}
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests-context.xml
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests-context.xml
@@ -7,7 +7,16 @@

 	<int:syslog-to-map-transformer id="toMap" input-channel="toMapChannel" output-channel="out" />

+	<int:syslog-to-map-transformer id="withPollerContextLoads" input-channel="nullChannel" output-channel="nullChannel">
+		<int:poller fixed-delay="50000" />
+	</int:syslog-to-map-transformer>
+
+	<int:chain input-channel="toMapChannel" output-channel="out">
+		<int:syslog-to-map-transformer />
+	</int:chain>
+
 	<int:channel id="out">
 		<int:queue />
 	</int:channel>
+
 </beans>
--- 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
@@ -25,6 +25,8 @@ import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
 import org.springframework.util.StringUtils;

 /**
@@ -40,7 +42,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 	@Override
 	protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-				this.getInboundFileSynchronizerClassname());
+				this.getInboundFileSynchronizerClass());

 		synchronizerBuilder.addConstructorArgReference(element.getAttribute("session-factory"));

@@ -54,7 +56,8 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 		String remoteFileSeparator = element.getAttribute("remote-file-separator");
 		synchronizerBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "temporary-file-suffix");
-		this.configureFilter(synchronizerBuilder, element, parserContext);
+		FileParserUtils.configureFilter(synchronizerBuilder, element, parserContext,
+				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass());

 		// build the MessageSource
 		BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(this.getMessageSourceClassname());
@@ -75,52 +78,12 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 		return messageSourceBuilder.getBeanDefinition();
 	}

-	private void configureFilter(BeanDefinitionBuilder synchronizerBuilder, Element element, ParserContext parserContext) {
-		String filter = element.getAttribute("filter");
-		String fileNamePattern = element.getAttribute("filename-pattern");
-		String fileNameRegex = element.getAttribute("filename-regex");
-		boolean hasFilter = StringUtils.hasText(filter);
-		boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
-		boolean hasFileNameRegex = StringUtils.hasText(fileNameRegex);
-		if (hasFilter || hasFileNamePattern || hasFileNameRegex) {
-			int count = 0;
-			if (hasFilter) {
-				count++;
-			}
-			if (hasFileNamePattern) {
-				count++;
-			}
-			if (hasFileNameRegex) {
-				count++;
-			}
-			if (count != 1) {
-				parserContext.getReaderContext().error("at most one of 'filename-pattern', " +
-						"'filename-regex', or 'filter' is allowed on remote file inbound adapter", element);
-			}
-			if (hasFilter) {
-				synchronizerBuilder.addPropertyReference("filter", filter);
-			}
-			else if (hasFileNamePattern) {
-				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-						this.getSimplePatternFileListFilterClassname());
-				filterBuilder.addConstructorArgValue(fileNamePattern);
-				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
-			}
-			else if (hasFileNameRegex) {
-				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-						this.getRegexPatternFileListFilterClassname());
-				filterBuilder.addConstructorArgValue(fileNameRegex);
-				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
-			}
-		}
-	}
-
 	protected abstract String getMessageSourceClassname();

-	protected abstract String getInboundFileSynchronizerClassname();
+	protected abstract Class<? extends InboundFileSynchronizer> getInboundFileSynchronizerClass();

-	protected abstract String getSimplePatternFileListFilterClassname();
+	protected abstract Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass();

-	protected abstract String getRegexPatternFileListFilterClassname();
+	protected abstract Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass();

 }
--- 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
@@ -0,0 +1,76 @@
+/*
+ * 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.file.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.BeanMetadataElement;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.RemoteFileOperations;
+import org.springframework.util.StringUtils;
+
+/**
+ * Abstract base class for parsing remote file streaming inbound channel adapters.
+ *
+ * @author Gary Russell
+ * @since 4.2.12
+ */
+public abstract class AbstractRemoteFileStreamingInboundChannelAdapterParser
+		extends AbstractPollingInboundChannelAdapterParser {
+
+	@Override
+	protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
+		BeanDefinition templateDefinition = FileParserUtils.parseRemoteFileTemplate(element, parserContext, false,
+				getTemplateClass());
+
+		BeanDefinitionBuilder messageSourceBuilder =
+				BeanDefinitionBuilder.genericBeanDefinition(getMessageSourceClass());
+		messageSourceBuilder.addConstructorArgValue(templateDefinition);
+
+		BeanDefinition expressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression(
+				"remote-directory", "remote-directory-expression", parserContext, element, false);
+		if (expressionDef != null) {
+			messageSourceBuilder.addPropertyValue("remoteDirectoryExpression", expressionDef);
+		}
+
+		String remoteFileSeparator = element.getAttribute("remote-file-separator");
+		messageSourceBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
+		FileParserUtils.configureFilter(messageSourceBuilder, element, parserContext,
+				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass());
+
+		String comparator = element.getAttribute("comparator");
+		if (StringUtils.hasText(comparator)) {
+			messageSourceBuilder.addConstructorArgReference(comparator);
+		}
+		return messageSourceBuilder.getBeanDefinition();
+	}
+
+	protected abstract Class<? extends RemoteFileOperations<?>> getTemplateClass();
+
+	protected abstract Class<? extends MessageSource<?>> getMessageSourceClass();
+
+	protected abstract Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass();
+
+	protected abstract Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass();
+
+}
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java
@@ -22,6 +22,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.file.DefaultFileNameGenerator;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.RemoteFileOperations;
 import org.springframework.util.StringUtils;

@@ -88,4 +89,43 @@ public final class FileParserUtils {
 		return templateBuilder.getBeanDefinition();
 	}

+	static void configureFilter(BeanDefinitionBuilder synchronizerBuilder, Element element, ParserContext parserContext,
+			Class<? extends FileListFilter<?>> patternClass, Class<? extends FileListFilter<?>> regexClass) {
+		String filter = element.getAttribute("filter");
+		String fileNamePattern = element.getAttribute("filename-pattern");
+		String fileNameRegex = element.getAttribute("filename-regex");
+		boolean hasFilter = StringUtils.hasText(filter);
+		boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
+		boolean hasFileNameRegex = StringUtils.hasText(fileNameRegex);
+		if (hasFilter || hasFileNamePattern || hasFileNameRegex) {
+			int count = 0;
+			if (hasFilter) {
+				count++;
+			}
+			if (hasFileNamePattern) {
+				count++;
+			}
+			if (hasFileNameRegex) {
+				count++;
+			}
+			if (count != 1) {
+				parserContext.getReaderContext().error("at most one of 'filename-pattern', " +
+						"'filename-regex', or 'filter' is allowed on remote file inbound adapter", element);
+			}
+			if (hasFilter) {
+				synchronizerBuilder.addPropertyReference("filter", filter);
+			}
+			else if (hasFileNamePattern) {
+				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(patternClass);
+				filterBuilder.addConstructorArgValue(fileNamePattern);
+				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
+			}
+			else if (hasFileNameRegex) {
+				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(regexClass);
+				filterBuilder.addConstructorArgValue(fileNameRegex);
+				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
+			}
+		}
+	}
+
 }
--- 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
@@ -141,7 +141,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F> extends Abstra
 						.build();
 			}
 			catch (IOException e) {
-				return new MessagingException("IOException when retrieving " + remotePath, e);
+				throw new MessagingException("IOException when retrieving " + remotePath, e);
 			}
 		}
 		return null;
--- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd
+++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd
@@ -727,15 +727,11 @@ Only files matching this regular expression will be picked up by this adapter.
 	</xsd:simpleType>

 	<xsd:attributeGroup name="remoteOutboundAttributeGroup">
-		<xsd:attribute name="remote-directory-expression"
-			type="xsd:string">
+		<xsd:attribute name="charset" type="xsd:string" default="UTF-8">
 			<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'");
+					Allows you to specify Charset (e.g., US-ASCII, ISO-8859-1, UTF-8). [UTF-8] is default -
+					used when converting String payloads to bytes.
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
--- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java
+++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java
@@ -106,7 +106,7 @@ public abstract class RemoteFileTestSupport {
 		recursiveDelete(new File(localTemporaryFolder.getRoot(), "localTarget"));
 		this.targetLocalDirectory = localTemporaryFolder.newFolder("localTarget");

-		File file = new File(this.sourceRemoteDirectory, " " + prefix + "Source1.txt");
+		File file = new File(this.sourceRemoteDirectory, prefix + "Source1.txt");
 		file.createNewFile();
 		FileOutputStream fos = new FileOutputStream(file);
 		fos.write("source1".getBytes());
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
@@ -17,6 +17,8 @@
 package org.springframework.integration.ftp.config;

 import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
@@ -37,18 +39,18 @@ public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundCha
 	}

 	@Override
-	protected String getInboundFileSynchronizerClassname() {
-		return FtpInboundFileSynchronizer.class.getName();
+	protected Class<? extends InboundFileSynchronizer> getInboundFileSynchronizerClass() {
+		return FtpInboundFileSynchronizer.class;
 	}

 	@Override
-	protected String getSimplePatternFileListFilterClassname() {
-		return FtpSimplePatternFileListFilter.class.getName();
+	protected Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass() {
+		return FtpSimplePatternFileListFilter.class;
 	}

 	@Override
-	protected String getRegexPatternFileListFilterClassname() {
-		return FtpRegexPatternFileListFilter.class.getName();
+	protected Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass() {
+		return FtpRegexPatternFileListFilter.class;
 	}

 }
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java
@@ -33,6 +33,8 @@ public class FtpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 	@Override
 	public void init() {
 		registerBeanDefinitionParser("inbound-channel-adapter", new FtpInboundChannelAdapterParser());
+		registerBeanDefinitionParser("inbound-streaming-channel-adapter",
+				new FtpStreamingInboundChannelAdapterParser());
 		registerBeanDefinitionParser("outbound-channel-adapter", new FtpOutboundChannelAdapterParser());
 		registerBeanDefinitionParser("outbound-gateway", new FtpOutboundGatewayParser());
 	}
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java
@@ -0,0 +1,55 @@
+/*
+ * 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.ftp.config;
+
+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;
+import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
+import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
+import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
+import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
+
+/**
+ * @author Gary Russell
+ * @since 4.2.12
+ *
+ */
+public class FtpStreamingInboundChannelAdapterParser extends AbstractRemoteFileStreamingInboundChannelAdapterParser {
+
+	@Override
+	protected Class<? extends RemoteFileOperations<?>> getTemplateClass() {
+		return FtpRemoteFileTemplate.class;
+	}
+
+	@Override
+	protected Class<? extends MessageSource<?>> getMessageSourceClass() {
+		return FtpStreamingMessageSource.class;
+	}
+
+	@Override
+	protected Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass() {
+		return FtpSimplePatternFileListFilter.class;
+	}
+
+	@Override
+	protected Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass() {
+		return FtpRegexPatternFileListFilter.class;
+	}
+
+}
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java
@@ -0,0 +1,74 @@
+/*
+ * 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.ftp.inbound;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.ftp.session.FtpFileInfo;
+
+/**
+ * Message source for streaming FTP remote file contents.
+ *
+ * @author Gary Russell
+ * @since 4.2.12
+ *
+ */
+public class FtpStreamingMessageSource extends AbstractRemoteFileStreamingMessageSource<FTPFile> {
+
+	/**
+	 * Construct an instance with the supplied template.
+	 * @param template the template.
+	 */
+	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template) {
+		super(template, null);
+	}
+
+	/**
+	 * Construct an instance with the supplied template and comparator.
+	 * Note: the comparator is applied each time the remote directory is listed
+	 * which only occurs when the previous list is exhausted.
+	 * @param template the template.
+	 * @param comparator the comparator.
+	 */
+	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template,
+			Comparator<AbstractFileInfo<FTPFile>> comparator) {
+		super(template, comparator);
+	}
+
+	@Override
+	public String getComponentType() {
+		return "ftp:inbound-streaming-channel-adapter";
+	}
+
+	@Override
+	protected List<AbstractFileInfo<FTPFile>> asFileInfoList(Collection<FTPFile> files) {
+		List<AbstractFileInfo<FTPFile>> canonicalFiles = new ArrayList<AbstractFileInfo<FTPFile>>();
+		for (FTPFile file : files) {
+			canonicalFiles.add(new FtpFileInfo(file));
+		}
+		return canonicalFiles;
+	}
+
+}
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java
@@ -146,7 +146,11 @@ public class FtpSession implements Session<FTPFile> {
 	public void close() {
 		try {
 			if (this.readingRaw.get()) {
-				finalizeRaw();
+				if (!finalizeRaw()) {
+					if (this.logger.isWarnEnabled()) {
+						this.logger.warn("Finalize on readRaw() returned false for " + this);
+					}
+				}
 			}
 			this.client.disconnect();
 		}
--- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.2.xsd
+++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.2.xsd
@@ -22,7 +22,7 @@
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>
-				<xsd:extension base="base-ftp-adapter-type">
+				<xsd:extension base="base-outbound-adapter-type">
 					<xsd:all>
 						<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
 					</xsd:all>
@@ -73,38 +73,8 @@
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>
-				<xsd:extension base="base-ftp-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. This channel where messages will be sent
-								to 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="local-filename-generator-expression"
-						type="xsd:string">
+				<xsd:extension base="base-inbound-adapter-type">
+					<xsd:attribute name="local-filename-generator-expression" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
 								Allows you to provide a SpEL expression to
@@ -120,17 +90,6 @@
 							</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="comparator" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation><![CDATA[
@@ -139,22 +98,6 @@
 							]]></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:attribute name="local-filter" type="xsd:string">
 						<xsd:annotation>
 							<xsd:appinfo>
@@ -215,16 +158,29 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="remote-directory-expression"
-						type="xsd:string">
+					<xsd:attributeGroup ref="tempSuffixGroup" />
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="inbound-streaming-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures a 'SourcePollingChannelAdapter' Endpoint for the
+				'org.springframework.integration.ftp.inbound.FtpInboundStreamingMessageSource'.
+			</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>
-								Specify a SpEL expression which
-								will be used to evaluate the directory
-								path from where the files will be transferred
-								(e.g., "@someBean.fetchDirectory");
-								Mutually exclusive with 'remote-directory'.
-							</xsd:documentation>
+							<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
+								FTP server. The generic type of the Comparator must be 'FtpFileInfo'.
+							]]></xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
 				</xsd:extension>
@@ -240,7 +196,7 @@
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>
-				<xsd:extension base="base-ftp-adapter-type">
+				<xsd:extension base="base-outbound-adapter-type">
 					<xsd:all>
 						<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
 						<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
@@ -522,14 +478,24 @@
 		</xsd:complexType>
 	</xsd:element>

-	<xsd:complexType name="base-ftp-adapter-type">
+	<xsd:complexType name="base-inbound-adapter-type">
 		<xsd:complexContent>
 			<xsd:extension base="base-adapter-type">
-				<xsd:attribute name="remote-directory" type="xsd:string" use="optional">
+				<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 the remote directory path (e.g., "/remote/mytransfers")
-							Mutually exclusive with 'remote-directory-expression'.
+							Identifies channel attached to this adapter.
+							The channel to which messages will be sent
+							by this adapter.
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
@@ -537,7 +503,46 @@
 					use="optional">
 					<xsd:annotation>
 						<xsd:documentation>
-							Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers")
+							Not used.
+						</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>
@@ -552,10 +557,24 @@
 		</xsd:complexContent>
 	</xsd:complexType>

+	<xsd:complexType name="base-outbound-adapter-type">
+		<xsd:complexContent>
+			<xsd:extension base="base-adapter-type">
+				<xsd:attribute name="temporary-remote-directory" type="xsd:string" use="optional">
+					<xsd:annotation>
+						<xsd:documentation>
+							Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers")
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attributeGroup ref="tempSuffixGroup" />
+			</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:attribute name="session-factory" type="xsd:string" use="required">
 			<xsd:annotation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
@@ -569,17 +588,6 @@
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="temporary-file-suffix" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation>
-					Extension used when downloading files. We
-					change
-					it right after we know it's
-					downloaded.
-				</xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
-
 		<xsd:attribute name="remote-file-separator" type="xsd:string"
 			default="/">
 			<xsd:annotation>
@@ -590,7 +598,39 @@
 				</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:attributeGroup name="tempSuffixGroup">
+		<xsd:attribute name="temporary-file-suffix" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Extension used when downloading files. We change
+					it right after we know it's downloaded.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
 </xsd:schema>
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright 2015 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.ftp;
+
+import java.io.File;
+import java.util.Arrays;
+
+import org.apache.commons.net.ftp.FTPFile;
+import org.apache.ftpserver.FtpServer;
+import org.apache.ftpserver.FtpServerFactory;
+import org.apache.ftpserver.ftplet.Authentication;
+import org.apache.ftpserver.ftplet.AuthenticationFailedException;
+import org.apache.ftpserver.ftplet.FtpException;
+import org.apache.ftpserver.ftplet.User;
+import org.apache.ftpserver.ftplet.UserManager;
+import org.apache.ftpserver.listener.Listener;
+import org.apache.ftpserver.listener.ListenerFactory;
+import org.apache.ftpserver.usermanager.impl.BaseUser;
+import org.apache.ftpserver.usermanager.impl.ConcurrentLoginPermission;
+import org.apache.ftpserver.usermanager.impl.TransferRatePermission;
+import org.apache.ftpserver.usermanager.impl.WritePermission;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import org.springframework.integration.file.remote.RemoteFileTestSupport;
+import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
+
+/**
+ * Provides an embedded FTP Server for test cases.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @author David Turanski
+ * @since 4.2.12
+ */
+public class FtpTestSupport extends RemoteFileTestSupport {
+
+	private static volatile FtpServer server;
+
+	public String getTargetLocalDirectoryName() {
+		return targetLocalDirectory.getAbsolutePath() + File.separator;
+	}
+
+	@BeforeClass
+	public static void createServer() throws Exception {
+		FtpServerFactory serverFactory = new FtpServerFactory();
+		serverFactory.setUserManager(new TestUserManager(remoteTemporaryFolder.getRoot().getAbsolutePath()));
+
+		ListenerFactory factory = new ListenerFactory();
+		factory.setPort(0);
+		serverFactory.addListener("default", factory.createListener());
+
+		server = serverFactory.createServer();
+		server.start();
+
+		Listener listener = serverFactory.getListeners().values().iterator().next();
+		port = listener.getPort();
+	}
+
+
+	@AfterClass
+	public static void stopServer() throws Exception {
+		server.stop();
+	}
+
+	@Override
+	protected String prefix() {
+		return "ftp";
+	}
+
+	public static SessionFactory<FTPFile> sessionFactory() {
+		DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
+		sf.setHost("localhost");
+		sf.setPort(port);
+		sf.setUsername("foo");
+		sf.setPassword("foo");
+
+		return new CachingSessionFactory<FTPFile>(sf);
+	}
+
+	private static class TestUserManager implements UserManager {
+
+		private final BaseUser testUser;
+
+		private TestUserManager(String homeDirectory) {
+			this.testUser = new BaseUser();
+			this.testUser.setAuthorities(Arrays.asList(new ConcurrentLoginPermission(1024, 1024),
+					new WritePermission(),
+					new TransferRatePermission(1024, 1024)));
+			this.testUser.setHomeDirectory(homeDirectory);
+			this.testUser.setName("TEST_USER");
+		}
+
+
+		@Override
+		public User getUserByName(String s) throws FtpException {
+			return this.testUser;
+		}
+
+		@Override
+		public String[] getAllUserNames() throws FtpException {
+			return new String[] { "TEST_USER" };
+		}
+
+		@Override
+		public void delete(String s) throws FtpException {
+		}
+
+		@Override
+		public void save(User user) throws FtpException {
+		}
+
+		@Override
+		public boolean doesExist(String s) throws FtpException {
+			return true;
+		}
+
+		@Override
+		public User authenticate(Authentication authentication) throws AuthenticationFailedException {
+			return this.testUser;
+		}
+
+		@Override
+		public String getAdminName() throws FtpException {
+			return "admin";
+		}
+
+		@Override
+		public boolean isAdmin(String s) throws FtpException {
+			return s.equals("admin");
+		}
+
+	}
+
+}
--- 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
@@ -0,0 +1,48 @@
+<?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="http://www.springframework.org/schema/integration"
+	xmlns:int-ftp="http://www.springframework.org/schema/integration/ftp"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd">
+
+	<bean id="ftpSessionFactory"
+		class="org.springframework.integration.ftp.config.FtpStreamingInboundChannelAdapterParserTests.TestSessionFactoryBean"/>
+
+	<bean id="csf" class="org.springframework.integration.file.remote.session.CachingSessionFactory">
+		<constructor-arg ref="ftpSessionFactory"/>
+	</bean>
+
+	<int-ftp:inbound-streaming-channel-adapter id="ftpInbound"
+				channel="ftpChannel"
+				session-factory="csf"
+				auto-startup="false"
+				phase="23"
+				filename-pattern="*.txt"
+				remote-file-separator="X"
+				comparator="comparator"
+				remote-directory-expression="'foo/bar'">
+			<int:poller fixed-rate="1000" />
+	</int-ftp:inbound-streaming-channel-adapter>
+
+	<int:channel id="ftpChannel">
+		<int:queue/>
+	</int:channel>
+
+	<bean id="comparator" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="java.util.Comparator"/>
+	</bean>
+
+	<int-ftp:inbound-streaming-channel-adapter id="contextLoadsWithNoComparator"
+				channel="nullChannel"
+				session-factory="csf"
+				auto-startup="false"
+				phase="23"
+				filename-pattern="*.txt"
+				remote-file-separator="X"
+				remote-directory-expression="'foo/bar'">
+			<int:poller fixed-rate="1000" />
+	</int-ftp:inbound-streaming-channel-adapter>
+
+</beans>
--- 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
@@ -0,0 +1,99 @@
+/*
+ * 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.ftp.config;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
+import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
+import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
+import org.springframework.integration.ftp.session.FtpSession;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class FtpStreamingInboundChannelAdapterParserTests {
+
+	@Autowired
+	private SourcePollingChannelAdapter ftpInbound;
+
+	@Autowired
+	private MessageChannel ftpChannel;
+
+	@Autowired
+	private CachingSessionFactory<?> csf;
+
+	@Test
+	public void testFtpInboundChannelAdapterComplete() throws Exception {
+		assertFalse(TestUtils.getPropertyValue(this.ftpInbound, "autoStartup", Boolean.class));
+		assertEquals("ftpInbound", this.ftpInbound.getComponentName());
+		assertEquals("ftp:inbound-streaming-channel-adapter", this.ftpInbound.getComponentType());
+		assertSame(this.ftpChannel, TestUtils.getPropertyValue(this.ftpInbound, "outputChannel"));
+		FtpStreamingMessageSource source = TestUtils.getPropertyValue(ftpInbound, "source",
+				FtpStreamingMessageSource.class);
+
+		assertNotNull(TestUtils.getPropertyValue(source, "comparator"));
+		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"));
+	}
+
+	public static class TestSessionFactoryBean implements FactoryBean<DefaultFtpSessionFactory> {
+
+		@Override
+		public DefaultFtpSessionFactory getObject() throws Exception {
+			DefaultFtpSessionFactory factory = mock(DefaultFtpSessionFactory.class);
+			FtpSession session = mock(FtpSession.class);
+			when(factory.getSession()).thenReturn(session);
+			return factory;
+		}
+
+		@Override
+		public Class<?> getObjectType() {
+			return DefaultFtpSessionFactory.class;
+		}
+
+		@Override
+		public boolean isSingleton() {
+			return true;
+		}
+	}
+
+}
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
@@ -0,0 +1,128 @@
+/*
+ * 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.ftp.inbound;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+
+import java.io.InputStream;
+
+import org.apache.commons.net.ftp.FTPFile;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+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.Transformer;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.ftp.FtpTestSupport;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.transformer.StreamTransformer;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 4.2.12
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class FtpStreamingMessageSourceTests extends FtpTestSupport {
+
+	@Autowired
+	public PollableChannel data;
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testAllContents() {
+		Message<byte[]> received = (Message<byte[]>) this.data.receive(10000);
+		assertNotNull(received);
+		assertThat(new String(received.getPayload()), equalTo("source1"));
+		received = (Message<byte[]>) this.data.receive(10000);
+		assertNotNull(received);
+		assertThat(new String(received.getPayload()), equalTo("source2"));
+		assertNull(this.data.receive(0));
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+		@Bean
+		public QueueChannel data() {
+			return new QueueChannel();
+		}
+
+		@Bean(name = PollerMetadata.DEFAULT_POLLER)
+		public PollerMetadata defaultPoller() {
+			PollerMetadata pollerMetadata = new PollerMetadata();
+			pollerMetadata.setTrigger(new PeriodicTrigger(500));
+			pollerMetadata.setMaxMessagesPerPoll(2000);
+			return pollerMetadata;
+		}
+
+		@Bean
+		public SubscribableChannel stream() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		@InboundChannelAdapter("stream")
+		public MessageSource<InputStream> ftpMessageSource() {
+			FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);
+			messageSource.setRemoteDirectory("ftpSource/");
+			messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "foo"));
+			return messageSource;
+		}
+
+		@Bean
+		@Transformer(inputChannel = "stream", outputChannel = "data")
+		public org.springframework.integration.transformer.Transformer transformer() {
+			return new StreamTransformer();
+		}
+
+		@Bean
+		public FtpRemoteFileTemplate template() {
+			return new FtpRemoteFileTemplate(ftpSessionFactory());
+		}
+
+		@Bean
+		public SessionFactory<FTPFile> ftpSessionFactory() {
+			return FtpStreamingMessageSourceTests.sessionFactory();
+		}
+
+	}
+
+}
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests-context.xml
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests-context.xml
@@ -1,15 +0,0 @@
-<?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-ftp="http://www.springframework.org/schema/integration/ftp"
-	   xmlns:int="http://www.springframework.org/schema/integration"
-	   xsi:schemaLocation="http://www.springframework.org/schema/integration/ftp
-	 http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
-
-	<bean id="ftpServer" class="org.springframework.integration.ftp.TestFtpServer">
-		<constructor-arg value="FtpRemoteFileTemplateTests"/>
-	</bean>
-
-</beans>
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java
@@ -29,12 +29,12 @@ import java.util.UUID;

 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
-import org.junit.After;
-import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;

 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.file.DefaultFileNameGenerator;
 import org.springframework.integration.file.remote.ClientCallbackWithoutResult;
@@ -42,7 +42,7 @@ import org.springframework.integration.file.remote.SessionCallback;
 import org.springframework.integration.file.remote.SessionCallbackWithoutResult;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
-import org.springframework.integration.ftp.TestFtpServer;
+import org.springframework.integration.ftp.FtpTestSupport;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
@@ -55,21 +55,11 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 */
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
-public class FtpRemoteFileTemplateTests {
-
-	@Autowired
-	private TestFtpServer ftpServer;
+public class FtpRemoteFileTemplateTests extends FtpTestSupport {

 	@Autowired
 	private SessionFactory<FTPFile> sessionFactory;

-	@Before
-	@After
-	public void setup() {
-		this.ftpServer.recursiveDelete(ftpServer.getTargetLocalDirectory());
-		this.ftpServer.recursiveDelete(ftpServer.getTargetFtpDirectory());
-	}
-
 	@Test
 	public void testINT3412AppendStatRmdir() throws IOException {
 		FtpRemoteFileTemplate template = new FtpRemoteFileTemplate(sessionFactory);
@@ -142,4 +132,14 @@ public class FtpRemoteFileTemplateTests {
 		newFile.delete();
 	}

+	@Configuration
+	public static class Config {
+
+		@Bean
+		public SessionFactory<FTPFile> ftpSessionFactory() {
+			return FtpRemoteFileTemplateTests.sessionFactory();
+		}
+
+	}
+
 }
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
@@ -17,6 +17,8 @@
 package org.springframework.integration.sftp.config;

 import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
 import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
@@ -37,18 +39,18 @@ public class SftpInboundChannelAdapterParser extends AbstractRemoteFileInboundCh
 	}

 	@Override
-	protected String getInboundFileSynchronizerClassname() {
-		return SftpInboundFileSynchronizer.class.getName();
+	protected Class<? extends InboundFileSynchronizer> getInboundFileSynchronizerClass() {
+		return SftpInboundFileSynchronizer.class;
 	}

 	@Override
-	protected String getSimplePatternFileListFilterClassname() {
-		return SftpSimplePatternFileListFilter.class.getName();
+	protected Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass() {
+		return SftpSimplePatternFileListFilter.class;
 	}

 	@Override
-	protected String getRegexPatternFileListFilterClassname() {
-		return SftpRegexPatternFileListFilter.class.getName();
+	protected Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass() {
+		return SftpRegexPatternFileListFilter.class;
 	}

 }
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java
@@ -32,6 +32,7 @@ public class SftpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 	@Override
 	public void init() {
 		registerBeanDefinitionParser("inbound-channel-adapter", new SftpInboundChannelAdapterParser());
+		registerBeanDefinitionParser("inbound-streaming-channel-adapter", new SftpStreamingInboundChannelAdapterParser());
 		registerBeanDefinitionParser("outbound-channel-adapter", new SftpOutboundChannelAdapterParser());
 		registerBeanDefinitionParser("outbound-gateway", new SftpOutboundGatewayParser());
 	}
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java
@@ -0,0 +1,55 @@
+/*
+ * 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.sftp.config;
+
+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;
+import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
+import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
+import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource;
+import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
+
+/**
+ * @author Gary Russell
+ * @since 4.2.12
+ *
+ */
+public class SftpStreamingInboundChannelAdapterParser extends AbstractRemoteFileStreamingInboundChannelAdapterParser {
+
+	@Override
+	protected Class<? extends RemoteFileOperations<?>> getTemplateClass() {
+		return SftpRemoteFileTemplate.class;
+	}
+
+	@Override
+	protected Class<? extends MessageSource<?>> getMessageSourceClass() {
+		return SftpStreamingMessageSource.class;
+	}
+
+	@Override
+	protected Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass() {
+		return SftpSimplePatternFileListFilter.class;
+	}
+
+	@Override
+	protected Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass() {
+		return SftpRegexPatternFileListFilter.class;
+	}
+
+}
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java
@@ -0,0 +1,74 @@
+/*
+ * 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.sftp.inbound;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+
+import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.sftp.session.SftpFileInfo;
+
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+
+/**
+ * Message source for streaming SFTP remote file contents.
+ *
+ * @author Gary Russell
+ * @since 4.2.12
+ *
+ */
+public class SftpStreamingMessageSource extends AbstractRemoteFileStreamingMessageSource<LsEntry> {
+
+	/**
+	 * Construct an instance with the supplied template.
+	 * @param template the template.
+	 */
+	public SftpStreamingMessageSource(RemoteFileTemplate<LsEntry> template) {
+		super(template, null);
+	}
+
+	/**
+	 * Construct an instance with the supplied template and comparator.
+	 * Note: the comparator is applied each time the remote directory is listed
+	 * which only occurs when the previous list is exhausted.
+	 * @param template the template.
+	 * @param comparator the comparator.
+	 */
+	public SftpStreamingMessageSource(RemoteFileTemplate<LsEntry> template,
+			Comparator<AbstractFileInfo<LsEntry>> comparator) {
+		super(template, comparator);
+	}
+
+	@Override
+	public String getComponentType() {
+		return "sftp:inbound-streaming-channel-adapter";
+	}
+
+	@Override
+	protected List<AbstractFileInfo<LsEntry>> asFileInfoList(Collection<LsEntry> files) {
+		List<AbstractFileInfo<LsEntry>> canonicalFiles = new ArrayList<AbstractFileInfo<LsEntry>>();
+		for (LsEntry file : files) {
+			canonicalFiles.add(new SftpFileInfo(file));
+		}
+		return canonicalFiles;
+	}
+
+}
--- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.2.xsd
+++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.2.xsd
@@ -22,7 +22,7 @@
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>
-				<xsd:extension base="base-sftp-adapter-type">
+				<xsd:extension base="base-outbound-adapter-type">
 					<xsd:all>
 						<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
 						<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
@@ -71,68 +71,16 @@

 	<xsd:element name="inbound-channel-adapter">
 		<xsd:annotation>
-			<xsd:documentation><![CDATA[
-                Builds an inbound-channel-adapter that synchronizes with a remote SFTP endpoint.
-            ]]></xsd:documentation>
+			<xsd:documentation>
+				Configures a 'SourcePollingChannelAdapter' Endpoint for the
+				'org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource'
+				that synchronizes with a remote SFTP endpoint.
+			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>
-				<xsd:extension base="base-sftp-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. This channel where messages will be sent
-								to by this adapter.
-							</xsd:documentation>
-						</xsd:annotation>
-					</xsd:attribute>
-					<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 will be determined by the java.io.File implementation of Comparable.
-					]]></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: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="local-filename-generator-expression"
-						type="xsd:string">
+				<xsd:extension base="base-inbound-adapter-type">
+					<xsd:attribute name="local-filename-generator-expression" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
 								Allows you to provide a SpEL expression to
@@ -148,15 +96,12 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="filename-regex" type="xsd:string">
+					<xsd:attribute name="comparator" 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:documentation><![CDATA[
+	Specify a Comparator to be used when ordering Files. If none is provided, the
+	order will be determined by the java.io.File implementation of Comparable.
+					]]></xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
 					<xsd:attribute name="local-filter" type="xsd:string">
@@ -219,16 +164,29 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="remote-directory-expression"
-						type="xsd:string">
+					<xsd:attributeGroup ref="tempSuffixGroup" />
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="inbound-streaming-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures a 'SourcePollingChannelAdapter' Endpoint for the
+				'org.springframework.integration.ftp.inbound.FtpInboundStreamingMessageSource'.
+			</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>
-								Specify a SpEL expression which
-								will be used to evaluate the directory
-								path from where the files will be transferred
-								(e.g., "@someBean.fetchDirectory").
-								Mutually exclusive with 'remote-directory'.
-							</xsd:documentation>
+							<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
+								SFTP server. The generic type of the Comparator must be 'SftpFileInfo'.
+							]]></xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
 				</xsd:extension>
@@ -244,7 +202,7 @@
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>
-				<xsd:extension base="base-sftp-adapter-type">
+				<xsd:extension base="base-outbound-adapter-type">
 					<xsd:all>
 						<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
 						<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
@@ -479,8 +437,7 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="auto-create-local-directory"
-						type="xsd:boolean">
+					<xsd:attribute name="auto-create-local-directory" type="xsd:boolean">
 						<xsd:annotation>
 							<xsd:documentation>
 								Tells this adapter if local directory must be
@@ -524,15 +481,63 @@
 		</xsd:complexType>
 	</xsd:element>

-	<xsd:complexType name="base-sftp-adapter-type">
+	<xsd:complexType name="base-inbound-adapter-type">
 		<xsd:complexContent>
 			<xsd:extension base="base-adapter-type">
-				<xsd:attribute name="remote-directory" type="xsd:string" use="optional">
+				<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>
-							Identifies the directory path (e.g.,
-							"/temp/mytransfers")
-								Mutually exclusive with 'remote-directory-expression'.
+							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>
@@ -540,7 +545,7 @@
 					use="optional">
 					<xsd:annotation>
 						<xsd:documentation>
-							Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers")
+							Not used.
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
@@ -557,10 +562,24 @@
 		</xsd:complexContent>
 	</xsd:complexType>

+	<xsd:complexType name="base-outbound-adapter-type">
+		<xsd:complexContent>
+			<xsd:extension base="base-adapter-type">
+				<xsd:attribute name="temporary-remote-directory" type="xsd:string" use="optional">
+					<xsd:annotation>
+						<xsd:documentation>
+							Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers")
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attributeGroup ref="tempSuffixGroup" />
+			</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:attribute name="session-factory" type="xsd:string" use="required">
 			<xsd:annotation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
@@ -574,16 +593,6 @@
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="temporary-file-suffix" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation>
-					Extension used when downloading files. We
-					change
-					it right after we know it's
-					downloaded.
-				</xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
 		<xsd:attribute name="remote-file-separator" type="xsd:string"
 			default="/">
 			<xsd:annotation>
@@ -594,7 +603,39 @@
 				</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:attributeGroup name="tempSuffixGroup">
+		<xsd:attribute name="temporary-file-suffix" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Extension used when downloading files. We change
+					it right after we know it's downloaded.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
 </xsd:schema>
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java
@@ -0,0 +1,97 @@
+/*
+ * 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.sftp;
+
+import java.io.File;
+import java.util.Collections;
+
+import org.apache.sshd.SshServer;
+import org.apache.sshd.common.NamedFactory;
+import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory;
+import org.apache.sshd.server.Command;
+import org.apache.sshd.server.PasswordAuthenticator;
+import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
+import org.apache.sshd.server.sftp.SftpSubsystem;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import org.springframework.integration.file.remote.RemoteFileTestSupport;
+import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
+
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+
+/**
+ * Provides an embedded SFTP Server for test cases.
+ *
+ * @author David Turanski
+ * @author Gary Russell
+ * @since 4.2.12
+ */
+public class SftpTestSupport extends RemoteFileTestSupport {
+
+	private static SshServer server;
+
+	public String getTargetLocalDirectoryName() {
+		return targetLocalDirectory.getAbsolutePath() + File.separator;
+	}
+
+	@Override
+	public String prefix() {
+		return "sftp";
+	}
+
+	@BeforeClass
+	public static void createServer() throws Exception {
+		server = SshServer.setUpDefaultServer();
+		server.setPasswordAuthenticator(new PasswordAuthenticator() {
+
+			@Override
+			public boolean authenticate(String username, String password,
+					org.apache.sshd.server.session.ServerSession session) {
+				return true;
+			}
+		});
+		server.setPort(0);
+		server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser"));
+		server.setSubsystemFactories(Collections.<NamedFactory<Command>>singletonList(new SftpSubsystem.Factory()));
+		server.setFileSystemFactory(new VirtualFileSystemFactory(remoteTemporaryFolder.getRoot().getAbsolutePath()));
+		server.start();
+		port = server.getPort();
+	}
+
+	public static SessionFactory<LsEntry> sessionFactory() {
+		DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
+		factory.setHost("localhost");
+		factory.setPort(port);
+		factory.setUser("foo");
+		factory.setPassword("foo");
+		factory.setAllowUnknownKeys(true);
+		return new CachingSessionFactory<LsEntry>(factory);
+	}
+
+	@AfterClass
+	public static void stopServer() throws Exception {
+		server.stop();
+		File hostkey = new File("hostkey.ser");
+		if (hostkey.exists()) {
+			hostkey.delete();
+		}
+	}
+
+}
--- 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
@@ -0,0 +1,48 @@
+<?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="http://www.springframework.org/schema/integration"
+	xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd">
+
+	<bean id="sftpSessionFactory"
+		class="org.springframework.integration.sftp.config.SftpStreamingInboundChannelAdapterParserTests$TestSessionFactoryBean"/>
+
+	<bean id="csf" class="org.springframework.integration.file.remote.session.CachingSessionFactory">
+		<constructor-arg ref="sftpSessionFactory"/>
+	</bean>
+
+	<int-sftp:inbound-streaming-channel-adapter id="sftpInbound"
+				channel="sftpChannel"
+				session-factory="csf"
+				auto-startup="false"
+				phase="23"
+				filename-pattern="*.txt"
+				remote-file-separator="X"
+				comparator="comparator"
+				remote-directory-expression="'foo/bar'">
+			<int:poller fixed-rate="1000" />
+	</int-sftp:inbound-streaming-channel-adapter>
+
+	<int:channel id="sftpChannel">
+		<int:queue/>
+	</int:channel>
+
+	<bean id="comparator" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="java.util.Comparator"/>
+	</bean>
+
+	<int-sftp:inbound-streaming-channel-adapter id="contextLoadsWithNoComparator"
+				channel="sftpChannel"
+				session-factory="csf"
+				auto-startup="false"
+				phase="23"
+				filename-pattern="*.txt"
+				remote-file-separator="X"
+				remote-directory-expression="'foo/bar'">
+			<int:poller fixed-rate="1000" />
+	</int-sftp:inbound-streaming-channel-adapter>
+
+</beans>
--- 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
@@ -0,0 +1,99 @@
+/*
+ * 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.sftp.config;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
+import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource;
+import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
+import org.springframework.integration.sftp.session.SftpSession;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class SftpStreamingInboundChannelAdapterParserTests {
+
+	@Autowired
+	private SourcePollingChannelAdapter sftpInbound;
+
+	@Autowired
+	private MessageChannel sftpChannel;
+
+	@Autowired
+	private CachingSessionFactory<?> csf;
+
+	@Test
+	public void testFtpInboundChannelAdapterComplete() throws Exception {
+		assertFalse(TestUtils.getPropertyValue(this.sftpInbound, "autoStartup", Boolean.class));
+		assertEquals("sftpInbound", this.sftpInbound.getComponentName());
+		assertEquals("sftp:inbound-streaming-channel-adapter", this.sftpInbound.getComponentType());
+		assertSame(this.sftpChannel, TestUtils.getPropertyValue(this.sftpInbound, "outputChannel"));
+		SftpStreamingMessageSource source = TestUtils.getPropertyValue(sftpInbound, "source",
+				SftpStreamingMessageSource.class);
+
+		assertNotNull(TestUtils.getPropertyValue(source, "comparator"));
+		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"));
+	}
+
+	public static class TestSessionFactoryBean implements FactoryBean<DefaultSftpSessionFactory> {
+
+		@Override
+		public DefaultSftpSessionFactory getObject() throws Exception {
+			DefaultSftpSessionFactory factory = mock(DefaultSftpSessionFactory.class);
+			SftpSession session = mock(SftpSession.class);
+			when(factory.getSession()).thenReturn(session);
+			return factory;
+		}
+
+		@Override
+		public Class<?> getObjectType() {
+			return DefaultSftpSessionFactory.class;
+		}
+
+		@Override
+		public boolean isSingleton() {
+			return true;
+		}
+	}
+
+}
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml
@@ -7,13 +7,15 @@
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
 		http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd">

+	<bean id="extraConfig" class="org.springframework.integration.sftp.inbound.RollbackLocalFilterTests$Config" />
+
 	<int-sftp:inbound-channel-adapter id="sftpAdapterAutoCreate"
 			session-factory="sftpSessionFactory"
 			channel="requestChannel"
 			remote-directory-expression="'/sftpSource'"
 			local-directory="file:local-test-dir/rollback"
 			auto-create-local-directory="true"
-			filename-pattern="sftpSource1.txt"
+			filename-pattern="sftpSource2.txt"
 			local-filter="acceptOnceFilter">
 		<int:poller fixed-rate="1000" max-messages-per-poll="2" error-channel="nullChannel">
 			<int:transactional synchronization-factory="syncFactory" />
@@ -34,6 +36,4 @@

 	<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager" />

-	<bean id="sftpServerConfig" class="org.springframework.integration.sftp.TestSftpServerConfig" />
-
 </beans>
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -30,10 +30,15 @@ import org.junit.Test;
 import org.junit.runner.RunWith;

 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.sftp.SftpTestSupport;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

+import com.jcraft.jsch.ChannelSftp.LsEntry;
+
 /**
 * @author Gary Russell
 * @author Artem Bilan
@@ -43,12 +48,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
-public class RollbackLocalFilterTests {
+public class RollbackLocalFilterTests extends SftpTestSupport {

 	@BeforeClass
 	@AfterClass
 	public static void clean() {
-		new File("local-test-dir/rollback/sftpSource1.txt").delete();
+		new File("local-test-dir/rollback/sftpSource2.txt").delete();
 	}

 	@Autowired
@@ -57,7 +62,7 @@ public class RollbackLocalFilterTests {
 	@Test
 	public void testRollback() throws Exception {
 		assertTrue(this.crash.getLatch().await(10, TimeUnit.SECONDS));
-		assertEquals("sftpSource1.txt", this.crash.getFile().getName());
+		assertEquals("sftpSource2.txt", this.crash.getFile().getName());
 	}

 	public static class Crash {
@@ -86,4 +91,13 @@ public class RollbackLocalFilterTests {
 		}
 	}

+	public static class Config {
+
+		@Bean
+		public SessionFactory<LsEntry> sftpSessionFactory() {
+			return RollbackLocalFilterTests.sessionFactory();
+		}
+
+	}
+
 }
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
@@ -0,0 +1,128 @@
+/*
+ * 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.sftp.inbound;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+
+import java.io.InputStream;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+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.Transformer;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.sftp.SftpTestSupport;
+import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
+import org.springframework.integration.transformer.StreamTransformer;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+
+/**
+ * @author Gary Russell
+ * @since 4.2.12
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class SftpStreamingMessageSourceTests extends SftpTestSupport {
+
+	@Autowired
+	public PollableChannel data;
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testAllContents() {
+		Message<byte[]> received = (Message<byte[]>) this.data.receive(10000);
+		assertNotNull(received);
+		assertThat(new String(received.getPayload()), equalTo("source1"));
+		received = (Message<byte[]>) this.data.receive(10000);
+		assertNotNull(received);
+		assertThat(new String(received.getPayload()), equalTo("source2"));
+		assertNull(this.data.receive(0));
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+		@Bean
+		public QueueChannel data() {
+			return new QueueChannel();
+		}
+
+		@Bean(name = PollerMetadata.DEFAULT_POLLER)
+		public PollerMetadata defaultPoller() {
+			PollerMetadata pollerMetadata = new PollerMetadata();
+			pollerMetadata.setTrigger(new PeriodicTrigger(500));
+			pollerMetadata.setMaxMessagesPerPoll(2000);
+			return pollerMetadata;
+		}
+
+		@Bean
+		@InboundChannelAdapter("stream")
+		public MessageSource<InputStream> ftpMessageSource() {
+			SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null);
+			messageSource.setRemoteDirectory("sftpSource/");
+			messageSource.setFilter(new AcceptOnceFileListFilter<LsEntry>());
+			return messageSource;
+		}
+
+		@Bean
+		public SubscribableChannel stream() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		@Transformer(inputChannel = "stream", outputChannel = "data")
+		public org.springframework.integration.transformer.Transformer transformer() {
+			return new StreamTransformer();
+		}
+
+		@Bean
+		public SftpRemoteFileTemplate template() {
+			return new SftpRemoteFileTemplate(ftpSessionFactory());
+		}
+
+		@Bean
+		public SessionFactory<LsEntry> ftpSessionFactory() {
+			return SftpStreamingMessageSourceTests.sessionFactory();
+		}
+
+	}
+
+}
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java
@@ -21,19 +21,20 @@ import static org.junit.Assert.assertTrue;

 import java.io.IOException;

-import org.junit.After;
-import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;

 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.file.DefaultFileNameGenerator;
 import org.springframework.integration.file.remote.ClientCallbackWithoutResult;
 import org.springframework.integration.file.remote.SessionCallback;
 import org.springframework.integration.file.remote.SessionCallbackWithoutResult;
 import org.springframework.integration.file.remote.session.Session;
-import org.springframework.integration.sftp.TestSftpServer;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.sftp.SftpTestSupport;
 import org.springframework.integration.sftp.TestSftpServerConfig;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
@@ -53,21 +54,11 @@ import com.jcraft.jsch.SftpException;
@ContextConfiguration(classes=TestSftpServerConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
-public class SftpRemoteFileTemplateTests {
-
-	@Autowired
-	private TestSftpServer sftpServer;
+public class SftpRemoteFileTemplateTests extends SftpTestSupport {

 	@Autowired
 	private DefaultSftpSessionFactory sessionFactory;

-	@Before
-	@After
-	public void setup() {
-		this.sftpServer.recursiveDelete(sftpServer.getTargetLocalDirectory());
-		this.sftpServer.recursiveDelete(sftpServer.getTargetSftpDirectory());
-	}
-
 	@Test
 	public void testINT3412AppendStatRmdir() {
 		SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(sessionFactory);
@@ -115,4 +106,14 @@ public class SftpRemoteFileTemplateTests {
 		assertFalse(template.exists("foo"));
 	}

+	@Configuration
+	public static class Config {
+
+		@Bean
+		public SessionFactory<LsEntry> ftpSessionFactory() {
+			return SftpRemoteFileTemplateTests.sessionFactory();
+		}
+
+	}
+
 }