DSL-(S)FTP-Namespace-Builder-Factory

**DO NOT MERGE YET**.

There is need to provide tests on the matter.

(S)FTP Tests

* Add test cases for (s)ftp components based on embedded (S)FtpServers
* Upgrade some dependencies
* Fix bug in the `IntegrationFlowBeanPostProcessor` for the registration internal (a result of `MessageSourceSpec`) as a bean in the AC

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java

@@ -31,6 +31,7 @@ import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
+import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec;
 import org.springframework.integration.dsl.support.MessageChannelReference;
@@ -132,12 +133,25 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
 					}
 					else if (component instanceof SourcePollingChannelAdapterSpec) {
 						SourcePollingChannelAdapterSpec spec = (SourcePollingChannelAdapterSpec) component;
-						String id = ((IntegrationComponentSpec) spec).getId();
 						SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1();
+						String id = ((IntegrationComponentSpec) spec).getId();
 						if (!StringUtils.hasText(id)) {
 							id = generateBeanName(pollingChannelAdapterFactoryBean);
 						}
 						registerComponent(pollingChannelAdapterFactoryBean, id);
+
+						MessageSource<?> messageSource = spec.get().getT2();
+						if (!this.beanFactory
+								.getBeansOfType(MessageSource.class, false, false)
+								.values()
+								.contains(messageSource)) {
+							String messageSourceId = id + ".source";
+							if (messageSource instanceof NamedComponent
+									&& ((NamedComponent) messageSource).getComponentName() != null) {
+								messageSourceId = ((NamedComponent) messageSource).getComponentName();
+							}
+							registerComponent(messageSource, messageSourceId);
+						}
 					}
 					else if (!this.beanFactory
 							.getBeansOfType(AopUtils.getTargetClass(component), false, false)

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

@@ -0,0 +1,152 @@
+/*
+ * 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.dsl.file;
+
+import java.lang.reflect.Constructor;
+import java.nio.charset.Charset;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.dsl.core.ComponentsRegistration;
+import org.springframework.integration.dsl.core.MessageHandlerSpec;
+import org.springframework.integration.file.DefaultFileNameGenerator;
+import org.springframework.integration.file.FileNameGenerator;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class FileTransferringMessageHandlerSpec<F, S extends FileTransferringMessageHandlerSpec<F, S>>
+		extends MessageHandlerSpec<S, FileTransferringMessageHandler<F>>
+		implements ComponentsRegistration {
+
+	private FileNameGenerator fileNameGenerator;
+
+	private DefaultFileNameGenerator defaultFileNameGenerator;
+
+	protected FileTransferringMessageHandlerSpec(SessionFactory<F> sessionFactory) {
+		this.target = new FileTransferringMessageHandler<F>(sessionFactory);
+	}
+
+	protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> remoteFileTemplate) {
+		this.target = new FileTransferringMessageHandler<F>(remoteFileTemplate);
+	}
+
+	@SuppressWarnings("unchecked")
+	protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> remoteFileTemplate, FileExistsMode fileExistsMode) {
+		Constructor<?> fileExistsModeConstructor =
+				ClassUtils.getConstructorIfAvailable(FileTransferringMessageHandler.class, RemoteFileTemplate.class,
+						FileExistsMode.class);
+		if (fileExistsModeConstructor == null) {
+			logger.warn("The 'FileExistsMode' constructor argument for the 'FileTransferringMessageHandler' is " +
+					"available since Spring Integration 4.1. Will be ignored for previous versions.");
+			this.target = new FileTransferringMessageHandler<F>(remoteFileTemplate);
+		}
+		else {
+			try {
+				this.target = (FileTransferringMessageHandler<F>) fileExistsModeConstructor.newInstance(remoteFileTemplate,
+								fileExistsMode);
+			}
+			catch (Exception e) {
+				throw new IllegalStateException(e);
+			}
+		}
+	}
+
+	public S autoCreateDirectory(boolean autoCreateDirectory) {
+		this.target.setAutoCreateDirectory(autoCreateDirectory);
+		return _this();
+	}
+
+	public S remoteFileSeparator(String remoteFileSeparator) {
+		this.target.setRemoteFileSeparator(remoteFileSeparator);
+		return _this();
+	}
+
+	public S remoteDirectory(String remoteDirectory) {
+		this.target.setRemoteDirectoryExpression(new LiteralExpression(remoteDirectory));
+		return _this();
+	}
+
+	public S remoteDirectoryExpression(String remoteDirectoryExpression) {
+		this.target.setRemoteDirectoryExpression(PARSER.parseExpression(remoteDirectoryExpression));
+		return _this();
+	}
+
+	public S temporaryRemoteDirectory(String temporaryRemoteDirectory) {
+		this.target.setTemporaryRemoteDirectoryExpression(new LiteralExpression(temporaryRemoteDirectory));
+		return _this();
+	}
+
+	public S temporaryRemoteDirectoryExpression(String temporaryRemoteDirectoryExpression) {
+		this.target.setTemporaryRemoteDirectoryExpression(PARSER.parseExpression(temporaryRemoteDirectoryExpression));
+		return _this();
+	}
+
+	public S useTemporaryFileName(boolean useTemporaryFileName) {
+		this.target.setUseTemporaryFileName(useTemporaryFileName);
+		return _this();
+	}
+
+	public S fileNameGenerator(FileNameGenerator fileNameGenerator) {
+		this.target.setFileNameGenerator(fileNameGenerator);
+		return _this();
+	}
+
+	public S fileNameGeneratorExpression(String fileNameGeneratorExpression) {
+		Assert.isNull(this.fileNameGenerator,
+				"'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive.");
+		this.defaultFileNameGenerator = new DefaultFileNameGenerator();
+		this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression);
+		return fileNameGenerator(this.defaultFileNameGenerator);
+	}
+
+	public S charset(String charset) {
+		this.target.setCharset(charset);
+		return _this();
+	}
+
+	public S charset(Charset charset) {
+		this.target.setCharset(charset.name());
+		return _this();
+	}
+
+	public S temporaryFileSuffix(String temporaryFileSuffix) {
+		this.target.setTemporaryFileSuffix(temporaryFileSuffix);
+		return _this();
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		if (this.defaultFileNameGenerator != null) {
+			return Collections.<Object>singletonList(this.defaultFileNameGenerator);
+		}
+		return Collections.<Object>emptyList();
+	}
+
+	@Override
+	protected FileTransferringMessageHandler<F> doGet() {
+		throw new UnsupportedOperationException();
+	}
+
+}

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

@@ -77,7 +77,7 @@ public class FileWritingMessageHandlerSpec
 				"'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive.");
 		this.defaultFileNameGenerator = new DefaultFileNameGenerator();
 		this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression);
-		return _this();
+		return fileNameGenerator(this.defaultFileNameGenerator);
 	}
 
 

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

@@ -0,0 +1,127 @@
+/*
+ * 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.dsl.file;
+
+import java.io.File;
+
+import org.springframework.integration.dsl.core.MessageHandlerSpec;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.filters.RegexPatternFileListFilter;
+import org.springframework.integration.file.filters.SimplePatternFileListFilter;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class RemoteFileOutboundGatewaySpec<F, S extends RemoteFileOutboundGatewaySpec<F, S>>
+		extends MessageHandlerSpec<S, AbstractRemoteFileOutboundGateway<F>> {
+
+	private FileListFilter<F> filter;
+
+	private FileListFilter<File> mputFilter;
+
+	protected RemoteFileOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<F> outboundGateway) {
+		this.target = outboundGateway;
+	}
+
+	public S options(String options) {
+		this.target.setOptions(options);
+		return _this();
+	}
+
+	public S options(AbstractRemoteFileOutboundGateway.Option... options) {
+		Assert.noNullElements(options);
+		StringBuilder optionsString = new StringBuilder();
+		for (AbstractRemoteFileOutboundGateway.Option option : options) {
+			optionsString.append(option.getOption()).append(" ");
+		}
+		this.target.setOptions(optionsString.toString());
+		return _this();
+	}
+
+	public S remoteFileSeparator(String remoteFileSeparator) {
+		this.target.setRemoteFileSeparator(remoteFileSeparator);
+		return _this();
+	}
+
+	public S localDirectory(java.io.File localDirectory) {
+		this.target.setLocalDirectory(localDirectory);
+		return _this();
+	}
+
+	public S localDirectoryExpression(String localDirectoryExpression) {
+		this.target.setLocalDirectoryExpression(PARSER.parseExpression(localDirectoryExpression));
+		return _this();
+	}
+
+	public S autoCreateLocalDirectory(boolean autoCreateLocalDirectory) {
+		this.target.setAutoCreateLocalDirectory(autoCreateLocalDirectory);
+		return _this();
+	}
+
+	public S temporaryFileSuffix(String temporaryFileSuffix) {
+		this.target.setTemporaryFileSuffix(temporaryFileSuffix);
+		return _this();
+	}
+
+	public S filter(FileListFilter<F> filter) {
+		Assert.isNull(this.filter,
+				"The 'filter' (" + this.filter + ") is already configured for the: " + this);
+		this.filter = filter;
+		this.target.setFilter(filter);
+		return _this();
+	}
+
+	public abstract S patternFileNameFilter(String pattern);
+
+	public abstract S regexFileNameFilter(String regex);
+
+	public S mputFilter(FileListFilter<File> filter) {
+		Assert.isNull(this.mputFilter,
+				"The 'filter' (" + this.mputFilter + ") is already configured for the: " + this);
+		this.mputFilter = filter;
+		this.target.setMputFilter(filter);
+		return _this();
+	}
+
+
+	public S patternMputFilter(String pattern) {
+		return mputFilter(new SimplePatternFileListFilter(pattern));
+	}
+
+	public S regexMpuFilter(String regex) {
+		return mputFilter(new RegexPatternFileListFilter(regex));
+	}
+
+
+	public S renameExpression(String expression) {
+		this.target.setRenameExpression(expression);
+		return _this();
+	}
+
+	public S localFilenameGeneratorExpression(String localFilenameGeneratorExpression) {
+		this.target.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameGeneratorExpression));
+		return _this();
+	}
+
+	@Override
+	protected AbstractRemoteFileOutboundGateway<F> doGet() {
+		throw new UnsupportedOperationException();
+	}
+
+}

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

@@ -0,0 +1,112 @@
+/*
+ * 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.dsl.file;
+
+import java.io.File;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.springframework.integration.dsl.core.ComponentsRegistration;
+import org.springframework.integration.dsl.core.MessageSourceSpec;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
+import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class RemoteInboundChannelAdapterSpec<F, S extends RemoteInboundChannelAdapterSpec<F, S, MS>,
+		MS extends AbstractInboundFileSynchronizingMessageSource<F>>
+		extends MessageSourceSpec<S, MS> implements ComponentsRegistration {
+
+	protected final AbstractInboundFileSynchronizer<F> synchronizer;
+
+	private FileListFilter<F> filter;
+
+	protected RemoteInboundChannelAdapterSpec(AbstractInboundFileSynchronizer<F> synchronizer) {
+		this.synchronizer = synchronizer;
+	}
+
+	public S autoCreateLocalDirectory(boolean autoCreateLocalDirectory) {
+		this.target.setAutoCreateLocalDirectory(autoCreateLocalDirectory);
+		return _this();
+	}
+
+	public S localDirectory(File localDirectory) {
+		this.target.setLocalDirectory(localDirectory);
+		return _this();
+	}
+
+	public S localFilter(FileListFilter<File> localFileListFilter) {
+		this.target.setLocalFilter(localFileListFilter);
+		return _this();
+	}
+
+	public S remoteFileSeparator(String remoteFileSeparator) {
+		this.synchronizer.setRemoteFileSeparator(remoteFileSeparator);
+		return _this();
+	}
+
+	public S localFilenameGeneratorExpression(String localFilenameGeneratorExpression) {
+		this.synchronizer.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameGeneratorExpression));
+		return _this();
+	}
+
+	public S temporaryFileSuffix(String temporaryFileSuffix) {
+		this.synchronizer.setTemporaryFileSuffix(temporaryFileSuffix);
+		return _this();
+	}
+
+	public S remoteDirectory(String remoteDirectory) {
+		this.synchronizer.setRemoteDirectory(remoteDirectory);
+		return _this();
+	}
+
+	public S filter(FileListFilter<F> filter) {
+		Assert.isNull(this.filter,
+				"The 'filter' (" + this.filter + ") is already configured for the: " + this);
+		this.filter = filter;
+		this.synchronizer.setFilter(filter);
+		return _this();
+	}
+
+	public abstract S patternFilter(String pattern);
+
+	public abstract S regexFilter(String regex);
+
+	public S deleteRemoteFiles(boolean deleteRemoteFiles) {
+		this.synchronizer.setDeleteRemoteFiles(deleteRemoteFiles);
+		return _this();
+	}
+
+	public S preserveTimestamp(boolean preserveTimestamp) {
+		this.synchronizer.setPreserveTimestamp(preserveTimestamp);
+		return _this();
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		return Collections.<Object>singletonList(this.synchronizer);
+	}
+
+	@Override
+	protected MS doGet() {
+		throw new UnsupportedOperationException();
+	}
+
+}

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

@@ -0,0 +1,71 @@
+/*
+ * 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.dsl.ftp;
+
+import java.util.Comparator;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+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;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class Ftp {
+
+	public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory<FTPFile> sessionFactory) {
+		return inboundAdapter(sessionFactory, null);
+	}
+
+	public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory<FTPFile> sessionFactory,
+			Comparator<java.io.File> receptionOrderComparator) {
+		return new FtpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator);
+	}
+
+	public static FtpMessageHandlerSpec outboundAdapter(SessionFactory<FTPFile> sessionFactory) {
+		return new FtpMessageHandlerSpec(sessionFactory);
+	}
+
+	public static FtpMessageHandlerSpec outboundAdapter(SessionFactory<FTPFile> sessionFactory,
+			FileExistsMode fileExistsMode) {
+		return outboundAdapter(new RemoteFileTemplate<FTPFile>(sessionFactory), fileExistsMode);
+	}
+
+	public static FtpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<FTPFile> remoteFileTemplate) {
+		return new FtpMessageHandlerSpec(remoteFileTemplate);
+	}
+
+	public static FtpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<FTPFile> remoteFileTemplate,
+			FileExistsMode fileExistsMode) {
+		return new FtpMessageHandlerSpec(remoteFileTemplate, fileExistsMode);
+	}
+
+	public static FtpOutboundGatewaySpec outboundGateway(SessionFactory<FTPFile> sessionFactory,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return outboundGateway(sessionFactory, command.getCommand(), expression);
+	}
+
+	public static FtpOutboundGatewaySpec outboundGateway(SessionFactory<FTPFile> sessionFactory,
+			String command, String expression) {
+		return new FtpOutboundGatewaySpec(new FtpOutboundGateway(sessionFactory, command, expression));
+	}
+
+}

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

@@ -0,0 +1,53 @@
+/*
+ * 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.dsl.ftp;
+
+import java.io.File;
+import java.util.Comparator;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec;
+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;
+
+/**
+ * @author Artem Bilan
+ */
+public class FtpInboundChannelAdapterSpec
+		extends RemoteInboundChannelAdapterSpec<FTPFile, FtpInboundChannelAdapterSpec,
+		FtpInboundFileSynchronizingMessageSource> {
+
+	FtpInboundChannelAdapterSpec(SessionFactory<FTPFile> sessionFactory, Comparator<File> comparator) {
+		super(new FtpInboundFileSynchronizer(sessionFactory));
+		this.target = new FtpInboundFileSynchronizingMessageSource(this.synchronizer, comparator);
+	}
+
+	@Override
+	public FtpInboundChannelAdapterSpec patternFilter(String pattern) {
+		return filter(new FtpSimplePatternFileListFilter(pattern));
+	}
+
+	@Override
+	public FtpInboundChannelAdapterSpec regexFilter(String regex) {
+		return filter(new FtpRegexPatternFileListFilter(regex));
+	}
+
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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.dsl.ftp;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.dsl.file.FileTransferringMessageHandlerSpec;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+
+/**
+ * @author Artem Bilan
+ */
+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-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpOutboundGatewaySpec.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.dsl.ftp;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.dsl.file.RemoteFileOutboundGatewaySpec;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
+import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
+
+/**
+ * @author Artem Bilan
+ */
+public class FtpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec<FTPFile, FtpOutboundGatewaySpec> {
+
+	FtpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<FTPFile> outboundGateway) {
+		super(outboundGateway);
+	}
+
+	@Override
+	public FtpOutboundGatewaySpec patternFileNameFilter(String pattern) {
+		return filter(new FtpSimplePatternFileListFilter(pattern));
+	}
+
+	@Override
+	public FtpOutboundGatewaySpec regexFileNameFilter(String regex) {
+		return filter(new FtpRegexPatternFileListFilter(regex));
+	}
+
+}

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

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

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

@@ -0,0 +1,71 @@
+/*
+ * 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.dsl.sftp;
+
+import java.util.Comparator;
+
+import com.jcraft.jsch.ChannelSftp;
+
+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;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class Sftp {
+
+	public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+		return inboundAdapter(sessionFactory, null);
+	}
+
+	public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			Comparator<java.io.File> receptionOrderComparator) {
+		return new SftpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator);
+	}
+
+	public static SftpMessageHandlerSpec outboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+		return new SftpMessageHandlerSpec(sessionFactory);
+	}
+
+	public static SftpMessageHandlerSpec outboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			FileExistsMode fileExistsMode) {
+		return outboundAdapter(new RemoteFileTemplate<ChannelSftp.LsEntry>(sessionFactory), fileExistsMode);
+	}
+
+	public static SftpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate) {
+		return new SftpMessageHandlerSpec(remoteFileTemplate);
+	}
+
+	public static SftpMessageHandlerSpec outboundAdapter(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate,
+			FileExistsMode fileExistsMode) {
+		return new SftpMessageHandlerSpec(remoteFileTemplate, fileExistsMode);
+	}
+
+	public static SftpOutboundGatewaySpec outboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return outboundGateway(sessionFactory, command.getCommand(), expression);
+	}
+
+	public static SftpOutboundGatewaySpec outboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			String command, String expression) {
+		return new SftpOutboundGatewaySpec(new SftpOutboundGateway(sessionFactory, command, expression));
+	}
+
+}

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

@@ -0,0 +1,53 @@
+/*
+ * 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.dsl.sftp;
+
+import java.io.File;
+import java.util.Comparator;
+
+import com.jcraft.jsch.ChannelSftp;
+
+import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec;
+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;
+
+/**
+ * @author Artem Bilan
+ */
+public class SftpInboundChannelAdapterSpec
+		extends RemoteInboundChannelAdapterSpec<ChannelSftp.LsEntry, SftpInboundChannelAdapterSpec,
+		SftpInboundFileSynchronizingMessageSource> {
+
+	SftpInboundChannelAdapterSpec(SessionFactory<ChannelSftp.LsEntry> sessionFactory, Comparator<File> comparator) {
+		super(new SftpInboundFileSynchronizer(sessionFactory));
+		this.target = new SftpInboundFileSynchronizingMessageSource(this.synchronizer, comparator);
+	}
+
+	@Override
+	public SftpInboundChannelAdapterSpec patternFilter(String pattern) {
+		return filter(new SftpSimplePatternFileListFilter(pattern));
+	}
+
+	@Override
+	public SftpInboundChannelAdapterSpec regexFilter(String regex) {
+		return filter(new SftpRegexPatternFileListFilter(regex));
+	}
+
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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.dsl.sftp;
+
+import com.jcraft.jsch.ChannelSftp;
+
+import org.springframework.integration.dsl.file.FileTransferringMessageHandlerSpec;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+
+/**
+ * @author Artem Bilan
+ */
+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-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.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.dsl.sftp;
+
+import com.jcraft.jsch.ChannelSftp;
+
+import org.springframework.integration.dsl.file.RemoteFileOutboundGatewaySpec;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
+import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
+
+/**
+ * @author Artem Bilan
+ */
+public class SftpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec<ChannelSftp.LsEntry, SftpOutboundGatewaySpec> {
+
+
+	SftpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<ChannelSftp.LsEntry> outboundGateway) {
+		super(outboundGateway);
+	}
+
+	@Override
+	public SftpOutboundGatewaySpec patternFileNameFilter(String pattern) {
+		return filter(new SftpSimplePatternFileListFilter(pattern));
+	}
+
+	@Override
+	public SftpOutboundGatewaySpec regexFileNameFilter(String regex) {
+		return filter(new SftpRegexPatternFileListFilter(regex));
+	}
+}

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

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

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java

@@ -16,7 +16,10 @@
 
 package org.springframework.integration.dsl.test;
 
-import static org.hamcrest.Matchers.*;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.endsWith;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.isOneOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -27,7 +30,6 @@ import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.io.FileInputStream;
-import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.charset.Charset;
@@ -42,12 +44,24 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
 
+import com.jcraft.jsch.ChannelSftp;
+import com.mongodb.MongoClient;
+import de.flapdoodle.embed.mongo.MongodExecutable;
+import de.flapdoodle.embed.mongo.MongodStarter;
+import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
+import de.flapdoodle.embed.mongo.config.Net;
+import de.flapdoodle.embed.mongo.distribution.Version;
+import de.flapdoodle.embed.process.runtime.Network;
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
+import org.apache.commons.net.ftp.FTPFile;
 import org.hamcrest.Matchers;
+import org.junit.After;
 import org.junit.AfterClass;
+import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -95,7 +109,9 @@ import org.springframework.integration.dsl.amqp.Amqp;
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.dsl.file.File;
+import org.springframework.integration.dsl.ftp.Ftp;
 import org.springframework.integration.dsl.jms.Jms;
+import org.springframework.integration.dsl.sftp.Sftp;
 import org.springframework.integration.dsl.support.Pollers;
 import org.springframework.integration.dsl.support.Transformers;
 import org.springframework.integration.endpoint.MethodInvokingMessageSource;
@@ -103,11 +119,15 @@ import org.springframework.integration.event.core.MessagingEvent;
 import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
 import org.springframework.integration.file.DefaultFileNameGenerator;
 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.ftp.session.DefaultFtpSessionFactory;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
 import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore;
 import org.springframework.integration.router.MethodInvokingRouter;
 import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
@@ -140,14 +160,6 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.StreamUtils;
 
-import com.mongodb.MongoClient;
-import de.flapdoodle.embed.mongo.MongodExecutable;
-import de.flapdoodle.embed.mongo.MongodStarter;
-import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
-import de.flapdoodle.embed.mongo.config.Net;
-import de.flapdoodle.embed.mongo.distribution.Version;
-import de.flapdoodle.embed.process.runtime.Network;
-
 /**
  * @author Artem Bilan
  * @author Tim Ysewyn
@@ -1102,6 +1114,160 @@ public class IntegrationFlowTests {
 	}
 
 
+	@Autowired
+	private TestFtpServer ftpServer;
+
+	@Autowired
+	private DefaultFtpSessionFactory ftpSessionFactory;
+
+	@Autowired
+	private TestSftpServer sftpServer;
+
+	@Autowired
+	private DefaultSftpSessionFactory sftpSessionFactory;
+
+	@Before
+	@After
+	public void setupRemoteFileServers() {
+		this.ftpServer.recursiveDelete(this.ftpServer.getTargetLocalDirectory());
+		this.ftpServer.recursiveDelete(this.ftpServer.getTargetFtpDirectory());
+		this.sftpServer.recursiveDelete(this.sftpServer.getTargetLocalDirectory());
+		this.sftpServer.recursiveDelete(this.sftpServer.getTargetSftpDirectory());
+	}
+
+	@Autowired
+	@Qualifier("ftpInboundResultChannel")
+	private PollableChannel ftpInboundResultChannel;
+
+	@Test
+	public void testFtpInboundFlow() {
+		Message<?> message = this.ftpInboundResultChannel.receive(1000);
+		assertNotNull(message);
+		Object payload = message.getPayload();
+		assertThat(payload, instanceOf(java.io.File.class));
+		java.io.File file = (java.io.File) payload;
+		assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("ftpTest"));
+
+		message = this.ftpInboundResultChannel.receive(1000);
+		assertNotNull(message);
+		file = (java.io.File) message.getPayload();
+		assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("ftpTest"));
+
+		this.controlBus.send("@ftpInboundAdapter.stop()");
+	}
+
+	@Autowired
+	@Qualifier("sftpInboundResultChannel")
+	private PollableChannel sftpInboundResultChannel;
+
+	@Test
+	public void testSftpInboundFlow() {
+		Message<?> message = this.sftpInboundResultChannel.receive(1000);
+		assertNotNull(message);
+		Object payload = message.getPayload();
+		assertThat(payload, instanceOf(java.io.File.class));
+		java.io.File file = (java.io.File) payload;
+		assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("sftpTest"));
+
+		message = this.sftpInboundResultChannel.receive(1000);
+		assertNotNull(message);
+		file = (java.io.File) message.getPayload();
+		assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
+		assertThat(file.getAbsolutePath(), containsString("sftpTest"));
+
+		this.controlBus.send("@sftpInboundAdapter.stop()");
+	}
+
+	@Autowired
+	@Qualifier("toFtpChannel")
+	private MessageChannel toFtpChannel;
+
+	@Test
+	public void testFtpOutboundFlow() {
+		String fileName = "foo.file";
+		this.toFtpChannel.send(MessageBuilder.withPayload("foo")
+				.setHeader(FileHeaders.FILENAME, fileName)
+				.build());
+
+		RemoteFileTemplate<FTPFile> template = new RemoteFileTemplate<>(this.ftpSessionFactory);
+		FTPFile[] files = template.execute(session ->
+				session.list(this.ftpServer.getTargetFtpDirectory().getName() + "/" + fileName));
+		assertEquals(1, files.length);
+		assertEquals(3, files[0].getSize());
+	}
+
+	@Autowired
+	@Qualifier("toSftpChannel")
+	private MessageChannel toSftpChannel;
+
+	@Test
+	public void testSftpOutboundFlow() {
+		String fileName = "foo.file";
+		this.toSftpChannel.send(MessageBuilder.withPayload("foo")
+				.setHeader(FileHeaders.FILENAME, fileName)
+				.build());
+
+		RemoteFileTemplate<ChannelSftp.LsEntry> template = new RemoteFileTemplate<>(this.sftpSessionFactory);
+		ChannelSftp.LsEntry[] files = template.execute(session ->
+				session.list(this.sftpServer.getTargetSftpDirectory().getName() + "/" + fileName));
+		assertEquals(1, files.length);
+		assertEquals(3, files[0].getAttrs().getSize());
+	}
+
+	@Autowired
+	@Qualifier("ftpMgetInputChannel")
+	private MessageChannel ftpMgetInputChannel;
+
+	@Autowired
+	@Qualifier("remoteFileOutputChannel")
+	private PollableChannel remoteFileOutputChannel;
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testFtpMgetFlow() {
+		String dir = "ftpSource/";
+		this.ftpMgetInputChannel.send(new GenericMessage<Object>(dir + "*"));
+		Message<?> result = this.remoteFileOutputChannel.receive(1000);
+		assertNotNull(result);
+		List<java.io.File> localFiles = (List<java.io.File>) result.getPayload();
+		// should have filtered ftpSource2.txt
+		assertEquals(2, localFiles.size());
+
+		for (java.io.File file : localFiles) {
+			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+					Matchers.containsString(dir));
+		}
+		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+				Matchers.containsString(dir + "subFtpSource"));
+	}
+
+
+	@Autowired
+	@Qualifier("sftpMgetInputChannel")
+	private MessageChannel sftpMgetInputChannel;
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testSftpMgetFlow() {
+		String dir = "sftpSource/";
+		this.sftpMgetInputChannel.send(new GenericMessage<Object>(dir + "*"));
+		Message<?> result = this.remoteFileOutputChannel.receive(1000);
+		assertNotNull(result);
+		List<java.io.File> localFiles = (List<java.io.File>) result.getPayload();
+		// should have filtered sftpSource2.txt
+		assertEquals(2, localFiles.size());
+
+		for (java.io.File file : localFiles) {
+			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+					Matchers.containsString(dir));
+		}
+		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+				Matchers.containsString(dir + "subSftpSource"));
+	}
+
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {
 
@@ -1144,7 +1310,7 @@ public class IntegrationFlowTests {
 
 		@Bean(name = PollerMetadata.DEFAULT_POLLER)
 		public PollerMetadata poller() {
-			return Pollers.fixedRate(500).get();
+			return Pollers.fixedRate(500).maxMessagesPerPoll(1).get();
 		}
 
 		@Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)
@@ -1214,6 +1380,96 @@ public class IntegrationFlowTests {
 					.<String, String>transform(String::toUpperCase)
 					.get();
 		}
+
+		@Autowired
+		private TestFtpServer ftpServer;
+
+		@Autowired
+		private DefaultFtpSessionFactory ftpSessionFactory;
+
+		@Autowired
+		private TestSftpServer sftpServer;
+
+		@Autowired
+		private DefaultSftpSessionFactory sftpSessionFactory;
+
+		@Bean
+		public IntegrationFlow ftpInboundFlow() {
+			return IntegrationFlows
+					.from(Ftp.inboundAdapter(this.ftpSessionFactory)
+									.preserveTimestamp(true)
+									.remoteDirectory("ftpSource")
+									.regexFilter(".*\\.txt$")
+									.localFilenameGeneratorExpression("#this.toUpperCase() + '.a'")
+									.localDirectory(this.ftpServer.getTargetLocalDirectory()),
+							e -> e.id("ftpInboundAdapter"))
+					.channel(MessageChannels.queue("ftpInboundResultChannel"))
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow sftpInboundFlow() {
+			return IntegrationFlows
+					.from(Sftp.inboundAdapter(this.sftpSessionFactory)
+									.preserveTimestamp(true)
+									.remoteDirectory("sftpSource")
+									.regexFilter(".*\\.txt$")
+									.localFilenameGeneratorExpression("#this.toUpperCase() + '.a'")
+									.localDirectory(this.sftpServer.getTargetLocalDirectory()),
+							e -> e.id("sftpInboundAdapter"))
+					.channel(MessageChannels.queue("sftpInboundResultChannel"))
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow ftpOutboundFlow() {
+			return IntegrationFlows.from("toFtpChannel")
+					.handle(Ftp.outboundAdapter(this.ftpSessionFactory)
+									.useTemporaryFileName(false)
+									.remoteDirectory(this.ftpServer.getTargetFtpDirectory().getName())
+					).get();
+		}
+
+		@Bean
+		public IntegrationFlow sftpOutboundFlow() {
+			return IntegrationFlows.from("toSftpChannel")
+					.handle(Sftp.outboundAdapter(this.sftpSessionFactory)
+									.useTemporaryFileName(false)
+									.remoteDirectory(this.sftpServer.getTargetSftpDirectory().getName())
+					).get();
+		}
+
+		@Bean
+		public PollableChannel remoteFileOutputChannel() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		public IntegrationFlow ftpMGetFlow() {
+			return IntegrationFlows.from("ftpMgetInputChannel")
+					.handle(Ftp.outboundGateway(this.ftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
+							"payload")
+							.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
+							.regexFileNameFilter("(subFtpSource|.*1.txt)")
+							.localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory")
+							.localFilenameGeneratorExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')"))
+					.channel(remoteFileOutputChannel())
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow sftpMGetFlow() {
+			return IntegrationFlows.from("sftpMgetInputChannel")
+					.handle(Sftp.outboundGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
+							"payload")
+							.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
+							.regexFileNameFilter("(subSftpSource|.*1.txt)")
+							.localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory")
+							.localFilenameGeneratorExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
+					.channel(remoteFileOutputChannel())
+					.get();
+		}
+
 	}
 
 	@Configuration

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestFtpServer.java

@@ -0,0 +1,253 @@
+/*
+ * 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.dsl.test;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+
+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.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.rules.TemporaryFolder;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
+import org.springframework.integration.test.util.SocketUtils;
+
+/**
+ * Embedded FTP Server for test cases; exposes an associated session factory
+ * as a @Bean.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ */
+@Configuration("ftpServer")
+public class TestFtpServer {
+
+	private final int ftpPort = SocketUtils.findAvailableServerSocket();
+
+	private final TemporaryFolder ftpFolder;
+
+	private final TemporaryFolder localFolder;
+
+	private volatile File ftpRootFolder;
+
+	private volatile File sourceFtpDirectory;
+
+	private volatile File targetFtpDirectory;
+
+	private volatile File sourceLocalDirectory;
+
+	private volatile File targetLocalDirectory;
+
+	private volatile FtpServer server;
+
+	public TestFtpServer() {
+		this.ftpFolder = new TemporaryFolder() {
+
+			@Override
+			public void create() throws IOException {
+				super.create();
+				ftpRootFolder = this.newFolder("ftpTest");
+				sourceFtpDirectory = new File(ftpRootFolder, "ftpSource");
+				sourceFtpDirectory.mkdir();
+				File file = new File(sourceFtpDirectory, "ftpSource1.txt");
+				file.createNewFile();
+				FileOutputStream fos = new FileOutputStream(file);
+				fos.write("source1".getBytes());
+				fos.close();
+				file = new File(sourceFtpDirectory, "ftpSource2.txt");
+				file.createNewFile();
+				fos = new FileOutputStream(file);
+				fos.write("source2".getBytes());
+				fos.close();
+
+				File subSourceFtpDirectory = new File(sourceFtpDirectory, "subFtpSource");
+				subSourceFtpDirectory.mkdir();
+				file = new File(subSourceFtpDirectory, "subFtpSource1.txt");
+				file.createNewFile();
+				fos = new FileOutputStream(file);
+				fos.write("subSource1".getBytes());
+				fos.close();
+
+				targetFtpDirectory = new File(ftpRootFolder, "ftpTarget");
+				targetFtpDirectory.mkdir();
+			}
+		};
+
+		this.localFolder  = new TemporaryFolder() {
+
+			@Override
+			public void create() throws IOException {
+				super.create();
+				File rootFolder = this.newFolder("ftpTest");
+				sourceLocalDirectory = new File(rootFolder, "localSource");
+				sourceLocalDirectory.mkdirs();
+				File file = new File(sourceLocalDirectory, "localSource1.txt");
+				file.createNewFile();
+				file = new File(sourceLocalDirectory, "localSource2.txt");
+				file.createNewFile();
+
+				File subSourceLocalDirectory = new File(sourceLocalDirectory, "subLocalSource");
+				subSourceLocalDirectory.mkdir();
+				file = new File(subSourceLocalDirectory, "subLocalSource1.txt");
+				file.createNewFile();
+
+				targetLocalDirectory = new File(rootFolder, "localTarget");
+				targetLocalDirectory.mkdir();
+			}
+		};
+	}
+
+	public File getSourceFtpDirectory() {
+		return sourceFtpDirectory;
+	}
+
+	public File getTargetFtpDirectory() {
+		return targetFtpDirectory;
+	}
+
+	public File getSourceLocalDirectory() {
+		return sourceLocalDirectory;
+	}
+
+	public File getTargetLocalDirectory() {
+		return targetLocalDirectory;
+	}
+
+	public String getTargetLocalDirectoryName() {
+		return targetLocalDirectory.getAbsolutePath() + File.separator;
+	}
+
+	@Bean
+	public DefaultFtpSessionFactory ftpSessionFactory() {
+		DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory();
+		factory.setHost("localhost");
+		factory.setPort(this.ftpPort);
+		factory.setUsername("foo");
+		factory.setPassword("foo");
+		return factory;
+	}
+
+	@PostConstruct
+	public void before() throws Throwable {
+		this.ftpFolder.create();
+		this.localFolder.create();
+
+		FtpServerFactory serverFactory = new FtpServerFactory();
+		serverFactory.setUserManager(new TestUserManager(this.ftpRootFolder.getAbsolutePath()));
+
+		ListenerFactory factory = new ListenerFactory();
+		factory.setPort(ftpPort);
+		serverFactory.addListener("default", factory.createListener());
+
+		server = serverFactory.createServer();
+		server.start();
+	}
+
+
+	@PreDestroy
+	public void after() {
+		this.server.stop();
+		this.ftpFolder.delete();
+		this.localFolder.delete();
+	}
+
+
+	public void recursiveDelete(File file) {
+		File[] files = file.listFiles();
+		if (files != null) {
+			for (File each : files) {
+				recursiveDelete(each);
+			}
+		}
+		if (!(file.equals(this.targetFtpDirectory) || file.equals(this.targetLocalDirectory))) {
+			file.delete();
+		}
+	}
+
+
+	private 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");
+		}
+
+	}
+
+}

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestSftpServer.java

@@ -0,0 +1,195 @@
+/*
+ * 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.dsl.test;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.sshd.SshServer;
+import org.apache.sshd.common.NamedFactory;
+import org.apache.sshd.common.file.FileSystemView;
+import org.apache.sshd.common.file.nativefs.NativeFileSystemFactory;
+import org.apache.sshd.common.file.nativefs.NativeFileSystemView;
+import org.apache.sshd.server.Command;
+import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
+import org.apache.sshd.server.sftp.SftpSubsystem;
+import org.junit.rules.TemporaryFolder;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
+import org.springframework.util.SocketUtils;
+
+/**
+ * @author Gary Russell
+ * @author Artem Bilan
+ */
+@Configuration("sftpServer")
+public class TestSftpServer implements InitializingBean, DisposableBean {
+
+	private final SshServer server = SshServer.setUpDefaultServer();
+
+	private final int port = SocketUtils.findAvailableTcpPort();
+
+	private final TemporaryFolder sftpFolder;
+
+	private final TemporaryFolder localFolder;
+
+	private volatile File sftpRootFolder;
+
+	private volatile File sourceSftpDirectory;
+
+	private volatile File targetSftpDirectory;
+
+	private volatile File sourceLocalDirectory;
+
+	private volatile File targetLocalDirectory;
+
+	public TestSftpServer() {
+		this.sftpFolder = new TemporaryFolder() {
+
+			@Override
+			public void create() throws IOException {
+				super.create();
+				sftpRootFolder = this.newFolder("sftpTest");
+				sourceSftpDirectory = new File(sftpRootFolder, "sftpSource");
+				sourceSftpDirectory.mkdir();
+				File file = new File(sourceSftpDirectory, "sftpSource1.txt");
+				file.createNewFile();
+				FileOutputStream fos = new FileOutputStream(file);
+				fos.write("source1".getBytes());
+				fos.close();
+				file = new File(sourceSftpDirectory, "sftpSource2.txt");
+				file.createNewFile();
+				fos = new FileOutputStream(file);
+				fos.write("source2".getBytes());
+				fos.close();
+
+				File subSourceFtpDirectory = new File(sourceSftpDirectory, "subSftpSource");
+				subSourceFtpDirectory.mkdir();
+				file = new File(subSourceFtpDirectory, "subSftpSource1.txt");
+				file.createNewFile();
+				fos = new FileOutputStream(file);
+				fos.write("subSource1".getBytes());
+				fos.close();
+
+				targetSftpDirectory = new File(sftpRootFolder, "sftpTarget");
+				targetSftpDirectory.mkdir();
+			}
+		};
+
+		this.localFolder = new TemporaryFolder() {
+
+			@Override
+			public void create() throws IOException {
+				super.create();
+				File rootFolder = this.newFolder("sftpTest");
+				sourceLocalDirectory = new File(rootFolder, "localSource");
+				sourceLocalDirectory.mkdirs();
+				File file = new File(sourceLocalDirectory, "localSource1.txt");
+				file.createNewFile();
+				file = new File(sourceLocalDirectory, "localSource2.txt");
+				file.createNewFile();
+
+				File subSourceLocalDirectory = new File(sourceLocalDirectory, "subLocalSource");
+				subSourceLocalDirectory.mkdir();
+				file = new File(subSourceLocalDirectory, "subLocalSource1.txt");
+				file.createNewFile();
+
+				targetLocalDirectory = new File(rootFolder, "localTarget");
+				targetLocalDirectory.mkdir();
+			}
+		};
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		this.sftpFolder.create();
+		this.localFolder.create();
+		this.server.setPasswordAuthenticator((username, password, session) -> true);
+		this.server.setPort(this.port);
+		this.server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser"));
+		SftpSubsystem.Factory sftp = new SftpSubsystem.Factory();
+		this.server.setSubsystemFactories(Arrays.<NamedFactory<Command>>asList(sftp));
+		this.server.setFileSystemFactory(new NativeFileSystemFactory() {
+
+			@Override
+			public FileSystemView createFileSystemView(org.apache.sshd.common.Session session) {
+				return new NativeFileSystemView(session.getUsername(), false) {
+
+					@Override
+					public String getVirtualUserDir() {
+						return sftpRootFolder.getAbsolutePath();
+					}
+				};
+			}
+
+		});
+		this.server.start();
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		this.server.stop();
+		this.sftpFolder.delete();
+		this.localFolder.delete();
+	}
+
+	public File getSourceLocalDirectory() {
+		return this.sourceLocalDirectory;
+	}
+
+	public File getTargetLocalDirectory() {
+		return this.targetLocalDirectory;
+	}
+
+	public String getTargetLocalDirectoryName() {
+		return this.targetLocalDirectory.getAbsolutePath() + File.separator;
+	}
+
+	public File getTargetSftpDirectory() {
+		return this.targetSftpDirectory;
+	}
+
+	public void recursiveDelete(File file) {
+		File[] files = file.listFiles();
+		if (files != null) {
+			for (File each : files) {
+				recursiveDelete(each);
+			}
+		}
+		if (!(file.equals(this.targetSftpDirectory) || file.equals(this.targetLocalDirectory))) {
+			file.delete();
+		}
+	}
+
+	@Bean
+	public DefaultSftpSessionFactory sftpSessionFactory() {
+		DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
+		factory.setHost("localhost");
+		factory.setPort(this.port);
+		factory.setUser("foo");
+		factory.setPassword("foo");
+		return factory;
+	}
+
+}