INT-4155: Port FTP DSL

JIRA: https://jira.spring.io/browse/INT-4155

- Add streaming support
- Rework tests to use dynamic flow registrations due to the test FTP server is now a test support class.

Javadoc fixes

Checkstyle fix

Polishing - PR Comments

Polishing and Port SFTP

spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileOutboundGatewaySpec.java

@@ -280,8 +280,22 @@ public abstract class RemoteFileOutboundGatewaySpec<F, S extends RemoteFileOutbo
 		return _this();
 	}
 
+	/**
+	 * Specify a simple pattern to match remote files (e.g. '*.txt').
+	 * @param pattern the pattern.
+	 * @return the spec.
+	 * @see org.springframework.integration.file.filters.AbstractSimplePatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
 	public abstract S patternFileNameFilter(String pattern);
 
+	/**
+	 * Specify a simple pattern to match remote files (e.g. '[0-9].*.txt').
+	 * @param regex the regex pattern.
+	 * @return the spec.
+	 * @see org.springframework.integration.file.filters.AbstractRegexPatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
 	public abstract S regexFileNameFilter(String regex);
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileStreamingInboundChannelAdapterSpec.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.file.dsl;
+
+import java.util.function.Function;
+
+import org.springframework.expression.Expression;
+import org.springframework.integration.dsl.MessageSourceSpec;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
+import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource;
+import org.springframework.messaging.Message;
+
+/**
+ * A {@link MessageSourceSpec} for an {@link AbstractInboundFileSynchronizingMessageSource}.
+ *
+ * @param <F> the target file type.
+ * @param <S> the target {@link RemoteFileStreamingInboundChannelAdapterSpec} implementation type.
+ * @param <MS> the target {@link AbstractInboundFileSynchronizingMessageSource} implementation type.
+ *
+ * @author Gary Russell
+ *
+ * @since 5.0
+ */
+public abstract class
+	RemoteFileStreamingInboundChannelAdapterSpec<F, S extends RemoteFileStreamingInboundChannelAdapterSpec<F, S, MS>,
+		MS extends AbstractRemoteFileStreamingMessageSource<F>>
+		extends MessageSourceSpec<S, MS> {
+
+	private CompositeFileListFilter<F> filter;
+
+	/**
+	 * Configure the file name path separator used by the remote system. Defaults to '/'.
+	 * @param remoteFileSeparator the remoteFileSeparator.
+	 * @return the spec.
+	 */
+	public S remoteFileSeparator(String remoteFileSeparator) {
+		this.target.setRemoteFileSeparator(remoteFileSeparator);
+		return _this();
+	}
+
+	/**
+	 * Specify the full path to the remote directory.
+	 * @param remoteDirectory the remoteDirectory.
+	 * @return the spec.
+	 * @see AbstractRemoteFileStreamingMessageSource#setRemoteDirectory(String)
+	 */
+	public S remoteDirectory(String remoteDirectory) {
+		this.target.setRemoteDirectory(remoteDirectory);
+		return _this();
+	}
+
+	/**
+	 * Specify an expression that evaluates to the full path to the remote directory.
+	 * @param remoteDirectoryExpression The remote directory expression.
+	 * @return the spec.
+	 */
+	public S remoteDirectory(Expression remoteDirectoryExpression) {
+		this.target.setRemoteDirectoryExpression(remoteDirectoryExpression);
+		return _this();
+	}
+
+	/**
+	 * Specify a function that is invoked to determine the full path to the remote directory.
+	 * @param remoteDirectoryFunction The remote directory function.
+	 * @return the spec.
+	 */
+	public S remoteDirectory(Function<Message<?>, String> remoteDirectoryFunction) {
+		this.target.setRemoteDirectoryExpression(new FunctionExpression<>(remoteDirectoryFunction));
+		return _this();
+	}
+
+	/**
+	 * Configure a {@link FileListFilter} to be applied to the remote files before
+	 * copying them.
+	 * @param filter the filter.
+	 * @return the spec.
+	 */
+	public S filter(FileListFilter<F> filter) {
+		if (this.filter == null) {
+			if (filter instanceof CompositeFileListFilter) {
+				this.filter = (CompositeFileListFilter<F>) filter;
+			}
+			else {
+				this.filter = new CompositeFileListFilter<F>();
+				this.filter.addFilter(filter);
+			}
+			this.target.setFilter(this.filter);
+		}
+		else {
+			this.filter.addFilter(filter);
+		}
+		return _this();
+	}
+
+	/**
+	 * Configure a simple pattern filter (e.g. '*.txt').
+	 * @param pattern the pattern.
+	 * @return the spec.
+	 * @see #filter(FileListFilter)
+	 */
+	public abstract S patternFilter(String pattern);
+
+	/**
+	 * Configure a regex pattern filter (e.g. '[0-9].*.txt').
+	 * @param regex the regex.
+	 * @return the spec.
+	 * @see #filter(FileListFilter)
+	 */
+	public abstract S regexFilter(String regex);
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java

@@ -57,6 +57,10 @@ public abstract class RemoteFileTestSupport {
 		return targetRemoteDirectory;
 	}
 
+	public String getTargetRemoteDirectoryName() {
+		return targetRemoteDirectory.getAbsolutePath() + File.separator;
+	}
+
 	public File getSourceLocalDirectory() {
 		return sourceLocalDirectory;
 	}
@@ -65,6 +69,10 @@ public abstract class RemoteFileTestSupport {
 		return targetLocalDirectory;
 	}
 
+	public String getTargetLocalDirectoryName() {
+		return targetLocalDirectory.getAbsolutePath() + File.separator;
+	}
+
 	/**
 	 * Default implementation creates the following folder structures:
 	 *

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/Ftp.java

@@ -0,0 +1,204 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import java.io.File;
+import java.util.Comparator;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.MessageSessionCallback;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.integration.ftp.gateway.FtpOutboundGateway;
+import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
+
+/**
+ * The factory for FTP components.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ */
+public final class Ftp {
+
+	/**
+	 * A {@link FtpInboundChannelAdapterSpec} factory for an inbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @return the spec.
+	 */
+	public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory<FTPFile> sessionFactory) {
+		return inboundAdapter(sessionFactory, null);
+	}
+
+	/**
+	 * A {@link FtpInboundChannelAdapterSpec} factory for an inbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @param receptionOrderComparator the comparator.
+	 * @return the spec.
+	 */
+	public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory<FTPFile> sessionFactory,
+			Comparator<File> receptionOrderComparator) {
+		return new FtpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator);
+	}
+
+	/**
+	 * A {@link FtpStreamingInboundChannelAdapterSpec} factory for an inbound channel
+	 * adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @return the spec.
+	 */
+	public static FtpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
+			RemoteFileTemplate<FTPFile> remoteFileTemplate) {
+		return inboundStreamingAdapter(remoteFileTemplate, null);
+	}
+
+	/**
+	 * A {@link FtpStreamingInboundChannelAdapterSpec} factory for an inbound channel
+	 * adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param receptionOrderComparator the comparator.
+	 * @return the spec.
+	 */
+	public static FtpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
+			RemoteFileTemplate<FTPFile> remoteFileTemplate,
+			Comparator<AbstractFileInfo<FTPFile>> receptionOrderComparator) {
+		return new FtpStreamingInboundChannelAdapterSpec(remoteFileTemplate, receptionOrderComparator);
+	}
+
+	/**
+	 * A {@link FtpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @return the spec.
+	 */
+	public static FtpMessageHandlerSpec outboundAdapter(SessionFactory<FTPFile> sessionFactory) {
+		return new FtpMessageHandlerSpec(sessionFactory);
+	}
+
+	/**
+	 * A {@link FtpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @param fileExistsMode the file exists mode.
+	 * @return the spec.
+	 */
+	public static FtpMessageHandlerSpec outboundAdapter(SessionFactory<FTPFile> sessionFactory,
+			FileExistsMode fileExistsMode) {
+		return outboundAdapter(new FtpRemoteFileTemplate(sessionFactory), fileExistsMode);
+	}
+
+	/**
+	 * A {@link FtpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @return the spec.
+	 */
+	public static FtpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<FTPFile> remoteFileTemplate) {
+		return new FtpMessageHandlerSpec(remoteFileTemplate);
+	}
+
+	/**
+	 * A {@link FtpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param fileExistsMode the file exists mode.
+	 * @return the spec.
+	 */
+	public static FtpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<FTPFile> remoteFileTemplate,
+			FileExistsMode fileExistsMode) {
+		return new FtpMessageHandlerSpec(remoteFileTemplate, fileExistsMode);
+	}
+
+	/**
+	 * Produce a {@link FtpOutboundGatewaySpec} based on the {@link SessionFactory},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the
+	 * remoteFilePath.
+	 * @param sessionFactory the {@link SessionFactory}.
+	 * @param command the command to perform on the FTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link FtpOutboundGatewaySpec}
+	 */
+	public static FtpOutboundGatewaySpec outboundGateway(SessionFactory<FTPFile> sessionFactory,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return outboundGateway(sessionFactory, command.getCommand(), expression);
+	}
+
+	/**
+	 * Produce a {@link FtpOutboundGatewaySpec} based on the {@link SessionFactory},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the
+	 * remoteFilePath.
+	 * @param sessionFactory the {@link SessionFactory}.
+	 * @param command the command to perform on the FTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link FtpOutboundGatewaySpec}
+	 * @see RemoteFileTemplate
+	 */
+	public static FtpOutboundGatewaySpec outboundGateway(SessionFactory<FTPFile> sessionFactory,
+			String command, String expression) {
+		return new FtpOutboundGatewaySpec(new FtpOutboundGateway(sessionFactory, command, expression));
+	}
+
+	/**
+	 * Produce a {@link FtpOutboundGatewaySpec} based on the {@link RemoteFileTemplate},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the
+	 * remoteFilePath.
+	 * @param remoteFileTemplate the {@link RemoteFileTemplate}.
+	 * @param command the command to perform on the FTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link FtpOutboundGatewaySpec}
+	 * @see RemoteFileTemplate
+	 */
+	public static FtpOutboundGatewaySpec outboundGateway(RemoteFileTemplate<FTPFile> remoteFileTemplate,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return outboundGateway(remoteFileTemplate, command.getCommand(), expression);
+	}
+
+	/**
+	 * Produce a {@link FtpOutboundGatewaySpec} based on the {@link RemoteFileTemplate},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the
+	 * remoteFilePath.
+	 * @param remoteFileTemplate the {@link RemoteFileTemplate}.
+	 * @param command the command to perform on the FTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link FtpOutboundGatewaySpec}
+	 * @see RemoteFileTemplate
+	 */
+	public static FtpOutboundGatewaySpec outboundGateway(RemoteFileTemplate<FTPFile> remoteFileTemplate,
+			String command, String expression) {
+		return new FtpOutboundGatewaySpec(new FtpOutboundGateway(remoteFileTemplate, command, expression));
+	}
+
+
+	/**
+	 * Produce a {@link FtpOutboundGatewaySpec} based on the
+	 * {@link MessageSessionCallback}.
+	 * @param sessionFactory the {@link SessionFactory} to connect to.
+	 * @param messageSessionCallback the {@link MessageSessionCallback} to perform SFTP
+	 * operation(s) with the {@code Message} context.
+	 * @return the {@link FtpOutboundGatewaySpec}
+	 * @see MessageSessionCallback
+	 */
+	public static FtpOutboundGatewaySpec outboundGateway(SessionFactory<FTPFile> sessionFactory,
+			MessageSessionCallback<FTPFile, ?> messageSessionCallback) {
+		return new FtpOutboundGatewaySpec(new FtpOutboundGateway(sessionFactory, messageSessionCallback));
+	}
+
+	private Ftp() {
+		super();
+	}
+
+}

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpInboundChannelAdapterSpec.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import java.io.File;
+import java.util.Comparator;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.file.dsl.RemoteFileInboundChannelAdapterSpec;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
+import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
+import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
+import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource;
+
+/**
+ * A {@link RemoteFileInboundChannelAdapterSpec} for a
+ * {@link FtpInboundFileSynchronizingMessageSource}.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class FtpInboundChannelAdapterSpec
+		extends RemoteFileInboundChannelAdapterSpec<FTPFile, FtpInboundChannelAdapterSpec,
+				FtpInboundFileSynchronizingMessageSource> {
+
+	FtpInboundChannelAdapterSpec(SessionFactory<FTPFile> sessionFactory, Comparator<File> comparator) {
+		super(new FtpInboundFileSynchronizer(sessionFactory));
+		this.target = new FtpInboundFileSynchronizingMessageSource(this.synchronizer, comparator);
+	}
+
+	/**
+	 * Specify a simple pattern to match remote files.
+	 * @param pattern the pattern.
+	 * @see FtpSimplePatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public FtpInboundChannelAdapterSpec patternFilter(String pattern) {
+		return filter(new FtpSimplePatternFileListFilter(pattern));
+	}
+
+	/**
+	 * Specify a regular expression to match remote files.
+	 * @param regex the expression.
+	 * @see FtpRegexPatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public FtpInboundChannelAdapterSpec regexFilter(String regex) {
+		return filter(new FtpRegexPatternFileListFilter(regex));
+	}
+
+}

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpMessageHandlerSpec.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.file.dsl.FileTransferringMessageHandlerSpec;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+
+/**
+ * A {@link FileTransferringMessageHandlerSpec} for FTP.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class FtpMessageHandlerSpec extends FileTransferringMessageHandlerSpec<FTPFile, FtpMessageHandlerSpec> {
+
+	FtpMessageHandlerSpec(SessionFactory<FTPFile> sessionFactory) {
+		super(sessionFactory);
+	}
+
+	FtpMessageHandlerSpec(RemoteFileTemplate<FTPFile> remoteFileTemplate) {
+		super(remoteFileTemplate);
+	}
+
+	FtpMessageHandlerSpec(RemoteFileTemplate<FTPFile> remoteFileTemplate, FileExistsMode fileExistsMode) {
+		super(remoteFileTemplate, fileExistsMode);
+	}
+
+}

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpOutboundGatewaySpec.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.file.dsl.RemoteFileOutboundGatewaySpec;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
+import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
+
+/**
+ * A {@link RemoteFileOutboundGatewaySpec} for FTP.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class FtpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec<FTPFile, FtpOutboundGatewaySpec> {
+
+	FtpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<FTPFile> outboundGateway) {
+		super(outboundGateway);
+	}
+
+	/**
+	 * @see FtpSimplePatternFileListFilter
+	 */
+	@Override
+	public FtpOutboundGatewaySpec patternFileNameFilter(String pattern) {
+		return filter(new FtpSimplePatternFileListFilter(pattern));
+	}
+
+	/**
+	 * @see FtpRegexPatternFileListFilter
+	 */
+	@Override
+	public FtpOutboundGatewaySpec regexFileNameFilter(String regex) {
+		return filter(new FtpRegexPatternFileListFilter(regex));
+	}
+
+}

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import java.util.Comparator;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.file.dsl.RemoteFileStreamingInboundChannelAdapterSpec;
+import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
+import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
+import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
+
+/**
+ * A {@link RemoteFileStreamingInboundChannelAdapterSpec} for a
+ * {@link FtpStreamingMessageSource}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ */
+public class FtpStreamingInboundChannelAdapterSpec
+		extends RemoteFileStreamingInboundChannelAdapterSpec<FTPFile, FtpStreamingInboundChannelAdapterSpec,
+				FtpStreamingMessageSource> {
+
+	FtpStreamingInboundChannelAdapterSpec(RemoteFileTemplate<FTPFile> remoteFileTemplate,
+			Comparator<AbstractFileInfo<FTPFile>> comparator) {
+		this.target = new FtpStreamingMessageSource(remoteFileTemplate, comparator);
+	}
+
+	/**
+	 * Specify a simple pattern to match remote files (e.g. '*.txt').
+	 * @param pattern the pattern.
+	 * @see FtpSimplePatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public FtpStreamingInboundChannelAdapterSpec patternFilter(String pattern) {
+		return filter(new FtpSimplePatternFileListFilter(pattern));
+	}
+
+	/**
+	 * Specify a regular expression to match remote files (e.g. '[0-9].*.txt').
+	 * @param regex the expression.
+	 * @see FtpRegexPatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public FtpStreamingInboundChannelAdapterSpec regexFilter(String regex) {
+		return filter(new FtpRegexPatternFileListFilter(regex));
+	}
+
+}

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides FTP Components for the Java DSL.
+ */
+package org.springframework.integration.ftp.dsl;

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java

@@ -16,7 +16,6 @@
 
 package org.springframework.integration.ftp;
 
-import java.io.File;
 import java.util.Arrays;
 
 import org.apache.commons.net.ftp.FTPFile;
@@ -53,10 +52,6 @@ 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();

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java

@@ -0,0 +1,185 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.isOneOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+
+import java.io.File;
+import java.io.InputStream;
+import java.util.List;
+import java.util.regex.Matcher;
+
+import org.apache.commons.net.ftp.FTPFile;
+import org.hamcrest.Matchers;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.Pollers;
+import org.springframework.integration.dsl.StandardIntegrationFlow;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.integration.dsl.context.IntegrationFlowRegistration;
+import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.integration.ftp.FtpTestSupport;
+import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class FtpTests extends FtpTestSupport {
+
+	@Autowired
+	private IntegrationFlowContext flowContext;
+
+	@Test
+	public void testFtpInboundFlow() {
+		QueueChannel out = new QueueChannel();
+		IntegrationFlow flow = IntegrationFlows.from(Ftp.inboundAdapter(sessionFactory())
+						.preserveTimestamp(true)
+						.remoteDirectory("ftpSource")
+						.regexFilter(".*\\.txt$")
+						.localFilename(f -> f.toUpperCase() + ".a")
+						.localDirectory(getTargetLocalDirectory()),
+				e -> e.id("ftpInboundAdapter").poller(Pollers.fixedDelay(100)))
+			.channel(out)
+			.get();
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		Message<?> message = out.receive(10_000);
+		assertNotNull(message);
+		Object payload = message.getPayload();
+		assertThat(payload, instanceOf(File.class));
+		File file = (File) payload;
+		assertThat(file.getName(), isOneOf(" FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("localTarget"));
+
+		message = out.receive(10_000);
+		assertNotNull(message);
+		file = (File) message.getPayload();
+		assertThat(file.getName(), isOneOf(" FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("localTarget"));
+
+		registration.destroy();
+	}
+
+	@Test
+	public void testFtpInboundStreamFlow() throws Exception {
+		QueueChannel out = new QueueChannel();
+		StandardIntegrationFlow flow = IntegrationFlows.from(
+				Ftp.inboundStreamingAdapter(new FtpRemoteFileTemplate(sessionFactory()))
+						.remoteDirectory("ftpSource")
+						.regexFilter(".*\\.txt$"),
+				e -> e.id("ftpInboundAdapter").poller(Pollers.fixedDelay(100)))
+			.channel(out)
+			.get();
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		Message<?> message = out.receive(10_000);
+		assertNotNull(message);
+		assertThat(message.getPayload(), instanceOf(InputStream.class));
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE), isOneOf(" ftpSource1.txt", "ftpSource2.txt"));
+		new IntegrationMessageHeaderAccessor(message).getCloseableResource().close();
+
+		message = out.receive(10_000);
+		assertNotNull(message);
+		assertThat(message.getPayload(), instanceOf(InputStream.class));
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE), isOneOf(" ftpSource1.txt", "ftpSource2.txt"));
+		new IntegrationMessageHeaderAccessor(message).getCloseableResource().close();
+
+		registration.destroy();
+	}
+
+	@Test
+	public void testFtpOutboundFlow() {
+		IntegrationFlow flow = f -> f
+			.handle(Ftp.outboundAdapter(sessionFactory(), FileExistsMode.FAIL)
+					.useTemporaryFileName(false)
+					.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
+					.remoteDirectory("ftpTarget"));
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		String fileName = "foo.file";
+		registration.getInputChannel().send(MessageBuilder.withPayload("foo")
+			.setHeader(FileHeaders.FILENAME, fileName)
+			.build());
+		RemoteFileTemplate<FTPFile> template = new RemoteFileTemplate<>(sessionFactory());
+		FTPFile[] files = template.execute(session ->
+				session.list(getTargetRemoteDirectory().getName() + "/" + fileName));
+		assertEquals(1, files.length);
+		assertEquals(3, files[0].getSize());
+
+		registration.destroy();
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testFtpMgetFlow() {
+		QueueChannel out = new QueueChannel();
+		IntegrationFlow flow = f -> f
+				.handle(Ftp.outboundGateway(sessionFactory(),
+						AbstractRemoteFileOutboundGateway.Command.MGET, "payload")
+							.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
+							.regexFileNameFilter("(subFtpSource|.*1.txt)")
+							.localDirectoryExpression("'" + getTargetLocalDirectoryName() + "' + #remoteDirectory")
+							.localFilenameExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')"))
+				.channel(out);
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		String dir = "ftpSource/";
+		registration.getInputChannel().send(new GenericMessage<>(dir + "*"));
+		Message<?> result = out.receive(10_000);
+		assertNotNull(result);
+		List<File> localFiles = (List<File>) result.getPayload();
+		// should have filtered ftpSource2.txt
+		assertEquals(2, localFiles.size());
+
+		for (File file : localFiles) {
+			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
+					Matchers.containsString(dir));
+		}
+		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
+				Matchers.containsString(dir + "subFtpSource"));
+
+		registration.destroy();
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+	}
+
+}

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/Sftp.java

@@ -0,0 +1,200 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import java.io.File;
+import java.util.Comparator;
+
+import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.MessageSessionCallback;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.integration.sftp.gateway.SftpOutboundGateway;
+import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
+
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+
+/**
+ * The factory for SFTP components.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ */
+public final class Sftp {
+
+	/**
+	 * An {@link SftpInboundChannelAdapterSpec} factory for an inbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @return the spec.
+	 */
+	public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+		return inboundAdapter(sessionFactory, null);
+	}
+
+	/**
+	 * An {@link SftpInboundChannelAdapterSpec} factory for an inbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @param receptionOrderComparator the comparator.
+	 * @return the spec.
+	 */
+	public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			Comparator<File> receptionOrderComparator) {
+		return new SftpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator);
+	}
+
+	/**
+	 * An {@link SftpStreamingInboundChannelAdapterSpec} factory for an inbound channel
+	 * adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @return the spec.
+	 */
+	public static SftpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
+			RemoteFileTemplate<LsEntry> remoteFileTemplate) {
+		return inboundStreamingAdapter(remoteFileTemplate, null);
+	}
+
+	/**
+	 * An {@link SftpStreamingInboundChannelAdapterSpec} factory for an inbound channel
+	 * adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param receptionOrderComparator the comparator.
+	 * @return the spec.
+	 */
+	public static SftpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
+			RemoteFileTemplate<LsEntry> remoteFileTemplate,
+			Comparator<AbstractFileInfo<LsEntry>> receptionOrderComparator) {
+		return new SftpStreamingInboundChannelAdapterSpec(remoteFileTemplate, receptionOrderComparator);
+	}
+
+	/**
+	 * An {@link SftpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @return the spec.
+	 */
+	public static SftpMessageHandlerSpec outboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+		return new SftpMessageHandlerSpec(sessionFactory);
+	}
+
+	/**
+	 * An {@link SftpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param sessionFactory the session factory.
+	 * @param fileExistsMode the file exists mode.
+	 * @return the spec.
+	 */
+	public static SftpMessageHandlerSpec outboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			FileExistsMode fileExistsMode) {
+		return outboundAdapter(new SftpRemoteFileTemplate(sessionFactory), fileExistsMode);
+	}
+
+	/**
+	 * An {@link SftpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @return the spec.
+	 */
+	public static SftpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate) {
+		return new SftpMessageHandlerSpec(remoteFileTemplate);
+	}
+
+	/**
+	 * An {@link SftpMessageHandlerSpec} factory for an outbound channel adapter spec.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param fileExistsMode the file exists mode.
+	 * @return the spec.
+	 */
+	public static SftpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate,
+			FileExistsMode fileExistsMode) {
+		return new SftpMessageHandlerSpec(remoteFileTemplate, fileExistsMode);
+	}
+
+	/**
+	 * Produce a {@link SftpOutboundGatewaySpec} based on the {@link SessionFactory},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the
+	 * remoteFilePath.
+	 * @param sessionFactory the {@link SessionFactory}.
+	 * @param command the command to perform on the FTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link SftpOutboundGatewaySpec}
+	 */
+	public static SftpOutboundGatewaySpec outboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return outboundGateway(sessionFactory, command.getCommand(), expression);
+	}
+
+	/**
+	 * Produce a {@link SftpOutboundGatewaySpec} based on the {@link SessionFactory},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the
+	 * remoteFilePath.
+	 * @param sessionFactory the {@link SessionFactory}.
+	 * @param command the command to perform on the FTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link SftpOutboundGatewaySpec}
+	 * @see RemoteFileTemplate
+	 */
+	public static SftpOutboundGatewaySpec outboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			String command, String expression) {
+		return new SftpOutboundGatewaySpec(new SftpOutboundGateway(sessionFactory, command, expression));
+	}
+
+	/**
+	 * Produce a {@link SftpOutboundGatewaySpec} based on the {@link RemoteFileTemplate},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the remoteFilePath.
+	 * @param remoteFileTemplate the {@link RemoteFileTemplate} to be based on.
+	 * @param command the command to perform on the SFTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link SftpOutboundGatewaySpec}
+	 * @see RemoteFileTemplate
+	 */
+	public static SftpOutboundGatewaySpec outboundGateway(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return outboundGateway(remoteFileTemplate, command.getCommand(), expression);
+	}
+
+	/**
+	 * Produce a {@link SftpOutboundGatewaySpec} based on the {@link RemoteFileTemplate},
+	 * {@link AbstractRemoteFileOutboundGateway.Command} and {@code expression} for the remoteFilePath.
+	 * @param remoteFileTemplate the {@link RemoteFileTemplate} to be based on.
+	 * @param command the command to perform on the SFTP.
+	 * @param expression the remoteFilePath SpEL expression.
+	 * @return the {@link SftpOutboundGatewaySpec}
+	 * @see RemoteFileTemplate
+	 */
+	public static SftpOutboundGatewaySpec outboundGateway(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate,
+			String command, String expression) {
+		return new SftpOutboundGatewaySpec(new SftpOutboundGateway(remoteFileTemplate, command, expression));
+	}
+
+	/**
+	 * Produce a {@link SftpOutboundGatewaySpec} based on the {@link MessageSessionCallback}.
+	 * @param sessionFactory the {@link SessionFactory} to connect to.
+	 * @param messageSessionCallback the {@link MessageSessionCallback} to perform SFTP operation(s)
+	 *                               with the {@code Message} context.
+	 * @return the {@link SftpOutboundGatewaySpec}
+	 * @see MessageSessionCallback
+	 */
+	public static SftpOutboundGatewaySpec outboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			MessageSessionCallback<ChannelSftp.LsEntry, ?> messageSessionCallback) {
+		return new SftpOutboundGatewaySpec(new SftpOutboundGateway(sessionFactory, messageSessionCallback));
+	}
+
+	private Sftp() {
+		super();
+	}
+}

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpInboundChannelAdapterSpec.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2014 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.dsl;
+
+import java.io.File;
+import java.util.Comparator;
+
+import org.springframework.integration.file.dsl.RemoteFileInboundChannelAdapterSpec;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
+import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
+import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
+import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource;
+
+import com.jcraft.jsch.ChannelSftp;
+
+/**
+ * A {@link RemoteFileInboundChannelAdapterSpec} for a {@link SftpInboundFileSynchronizingMessageSource}.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class SftpInboundChannelAdapterSpec
+		extends RemoteFileInboundChannelAdapterSpec<ChannelSftp.LsEntry, SftpInboundChannelAdapterSpec,
+				SftpInboundFileSynchronizingMessageSource> {
+
+	SftpInboundChannelAdapterSpec(SessionFactory<ChannelSftp.LsEntry> sessionFactory, Comparator<File> comparator) {
+		super(new SftpInboundFileSynchronizer(sessionFactory));
+		this.target = new SftpInboundFileSynchronizingMessageSource(this.synchronizer, comparator);
+	}
+
+	/**
+	 * @param pattern the Ant style pattern filter to use.
+	 * @see SftpSimplePatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public SftpInboundChannelAdapterSpec patternFilter(String pattern) {
+		return filter(new SftpSimplePatternFileListFilter(pattern));
+	}
+
+	/**
+	 * @param regex the RegExp pattern to use.
+	 * @see SftpRegexPatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public SftpInboundChannelAdapterSpec regexFilter(String regex) {
+		return filter(new SftpRegexPatternFileListFilter(regex));
+	}
+
+}

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpMessageHandlerSpec.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2014 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.dsl;
+
+import org.springframework.integration.file.dsl.FileTransferringMessageHandlerSpec;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+
+import com.jcraft.jsch.ChannelSftp;
+
+/**
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class SftpMessageHandlerSpec
+		extends FileTransferringMessageHandlerSpec<ChannelSftp.LsEntry, SftpMessageHandlerSpec> {
+
+	SftpMessageHandlerSpec(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+		super(sessionFactory);
+	}
+
+	SftpMessageHandlerSpec(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate) {
+		super(remoteFileTemplate);
+	}
+
+	SftpMessageHandlerSpec(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate, FileExistsMode fileExistsMode) {
+		super(remoteFileTemplate, fileExistsMode);
+	}
+
+}

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpOutboundGatewaySpec.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2014 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.dsl;
+
+import org.springframework.integration.file.dsl.RemoteFileOutboundGatewaySpec;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
+import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
+
+import com.jcraft.jsch.ChannelSftp;
+
+/**
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ */
+public class SftpOutboundGatewaySpec
+		extends RemoteFileOutboundGatewaySpec<ChannelSftp.LsEntry, SftpOutboundGatewaySpec> {
+
+
+	SftpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<ChannelSftp.LsEntry> outboundGateway) {
+		super(outboundGateway);
+	}
+
+	/**
+	 * @see SftpSimplePatternFileListFilter
+	 */
+	@Override
+	public SftpOutboundGatewaySpec patternFileNameFilter(String pattern) {
+		return filter(new SftpSimplePatternFileListFilter(pattern));
+	}
+
+	/**
+	 * @see SftpRegexPatternFileListFilter
+	 */
+	@Override
+	public SftpOutboundGatewaySpec regexFileNameFilter(String regex) {
+		return filter(new SftpRegexPatternFileListFilter(regex));
+	}
+
+}

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpStreamingInboundChannelAdapterSpec.java

@@ -0,0 +1,65 @@
+/*
+ * 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.dsl;
+
+import java.util.Comparator;
+
+import org.springframework.integration.file.dsl.RemoteFileStreamingInboundChannelAdapterSpec;
+import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
+import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
+import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource;
+
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+
+/**
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class SftpStreamingInboundChannelAdapterSpec
+		extends RemoteFileStreamingInboundChannelAdapterSpec<LsEntry, SftpStreamingInboundChannelAdapterSpec,
+			SftpStreamingMessageSource> {
+
+	SftpStreamingInboundChannelAdapterSpec(RemoteFileTemplate<LsEntry> remoteFileTemplate,
+			Comparator<AbstractFileInfo<LsEntry>> comparator) {
+		this.target = new SftpStreamingMessageSource(remoteFileTemplate, comparator);
+	}
+
+	/**
+	 * Specify a simple pattern to match remote files (e.g. '*.txt').
+	 * @param pattern the pattern.
+	 * @see SftpSimplePatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public SftpStreamingInboundChannelAdapterSpec patternFilter(String pattern) {
+		return filter(new SftpSimplePatternFileListFilter(pattern));	}
+
+	/**
+	 * Specify a regular expression to match remote files (e.g. '[0-9].*.txt').
+	 * @param regex the expression.
+	 * @see SftpRegexPatternFileListFilter
+	 * @see #filter(org.springframework.integration.file.filters.FileListFilter)
+	 */
+	@Override
+	public SftpStreamingInboundChannelAdapterSpec regexFilter(String regex) {
+		return filter(new SftpRegexPatternFileListFilter(regex));
+	}
+
+}

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides SFTP Components for the Java DSL.
+ */
+package org.springframework.integration.sftp.dsl;

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java

@@ -0,0 +1,207 @@
+/*
+ * Copyright 2014-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.dsl;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.isOneOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.InputStream;
+import java.util.List;
+import java.util.regex.Matcher;
+
+import org.hamcrest.Matchers;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.Pollers;
+import org.springframework.integration.dsl.StandardIntegrationFlow;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.integration.dsl.context.IntegrationFlowRegistration;
+import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.integration.sftp.SftpTestSupport;
+import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import com.jcraft.jsch.ChannelSftp;
+
+/**
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class SftpTests extends SftpTestSupport {
+
+	@Autowired
+	private IntegrationFlowContext flowContext;
+
+	@Test
+	public void testSftpInboundFlow() {
+		QueueChannel out = new QueueChannel();
+		IntegrationFlow flow = IntegrationFlows
+			.from(Sftp.inboundAdapter(sessionFactory())
+							.preserveTimestamp(true)
+							.remoteDirectory("sftpSource")
+							.regexFilter(".*\\.txt$")
+							.localFilenameExpression("#this.toUpperCase() + '.a'")
+							.localDirectory(getTargetLocalDirectory()),
+					e -> e.id("sftpInboundAdapter").poller(Pollers.fixedDelay(100)))
+			.channel(out)
+			.get();
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		Message<?> message = out.receive(10_000);
+		assertNotNull(message);
+		Object payload = message.getPayload();
+		assertThat(payload, instanceOf(File.class));
+		File file = (File) payload;
+		assertThat(file.getName(), isOneOf(" SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("localTarget"));
+
+		message = out.receive(10_000);
+		assertNotNull(message);
+		file = (File) message.getPayload();
+		assertThat(file.getName(), isOneOf(" SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("localTarget"));
+
+		registration.destroy();
+	}
+
+	@Test
+	public void testSftpInboundStreamFlow() throws Exception {
+		QueueChannel out = new QueueChannel();
+		StandardIntegrationFlow flow = IntegrationFlows.from(
+				Sftp.inboundStreamingAdapter(new SftpRemoteFileTemplate(sessionFactory()))
+						.remoteDirectory("sftpSource")
+						.regexFilter(".*\\.txt$"),
+				e -> e.id("sftpInboundAdapter").poller(Pollers.fixedDelay(100)))
+			.channel(out)
+			.get();
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		Message<?> message = out.receive(10_000);
+		assertNotNull(message);
+		assertThat(message.getPayload(), instanceOf(InputStream.class));
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE), isOneOf(" sftpSource1.txt", "sftpSource2.txt"));
+		new IntegrationMessageHeaderAccessor(message).getCloseableResource().close();
+
+		message = out.receive(10_000);
+		assertNotNull(message);
+		assertThat(message.getPayload(), instanceOf(InputStream.class));
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE), isOneOf("sftpSource1.txt", "sftpSource2.txt"));
+		new IntegrationMessageHeaderAccessor(message).getCloseableResource().close();
+
+		registration.destroy();
+	}
+
+	@Test
+	public void testSftpOutboundFlow() {
+		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sessionFactory(), FileExistsMode.FAIL)
+				.useTemporaryFileName(false)
+				.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
+				.remoteDirectory("sftpTarget"));
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		String fileName = "foo.file";
+		registration.getInputChannel().send(MessageBuilder.withPayload("foo")
+				.setHeader(FileHeaders.FILENAME, fileName)
+				.build());
+
+		RemoteFileTemplate<ChannelSftp.LsEntry> template = new RemoteFileTemplate<>(sessionFactory());
+		ChannelSftp.LsEntry[] files = template.execute(session ->
+				session.list(getTargetRemoteDirectory().getName() + "/" + fileName));
+		assertEquals(1, files.length);
+		assertEquals(3, files[0].getAttrs().getSize());
+
+		registration.destroy();
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testSftpMgetFlow() {
+		QueueChannel out = new QueueChannel();
+		IntegrationFlow flow = f -> f
+				.handle(Sftp.outboundGateway(sessionFactory(), AbstractRemoteFileOutboundGateway.Command.MGET,
+						"payload")
+						.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
+						.regexFileNameFilter("(subSftpSource|.*1.txt)")
+						.localDirectoryExpression("'" + getTargetLocalDirectoryName() + "' + #remoteDirectory")
+						.localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
+				.channel(out);
+		String dir = "sftpSource/";
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		registration.getInputChannel().send(new GenericMessage<>(dir + "*"));
+		Message<?> result = out.receive(10_000);
+		assertNotNull(result);
+		List<File> localFiles = (List<File>) result.getPayload();
+		// should have filtered sftpSource2.txt
+		assertEquals(2, localFiles.size());
+
+		for (File file : localFiles) {
+			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
+					Matchers.containsString(dir));
+		}
+		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
+				Matchers.containsString(dir + "subSftpSource"));
+
+		registration.destroy();
+	}
+
+	@Test
+	public void testSftpSessionCallback() {
+		QueueChannel out = new QueueChannel();
+		IntegrationFlow flow = f -> f
+				.<String>handle((p, h) -> new SftpRemoteFileTemplate(sessionFactory()).execute(s -> s.list(p)))
+				.channel(out);
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		registration.getInputChannel().send(new GenericMessage<>("sftpSource"));
+		Message<?> receive = out.receive(10_000);
+		assertNotNull(receive);
+		Object payload = receive.getPayload();
+		assertThat(payload, instanceOf(ChannelSftp.LsEntry[].class));
+
+		assertTrue(((ChannelSftp.LsEntry[]) payload).length > 0);
+
+		registration.destroy();
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+	}
+
+}