INT-3622: (S)FTP: Add MessageSessionCallback

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

INT-3622: Polishing

spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileOutboundGatewayParser.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.integration.file.config;
 
 import org.w3c.dom.Element;
@@ -53,8 +54,13 @@ public abstract class AbstractRemoteFileOutboundGatewayParser extends AbstractCo
 
 		builder.addConstructorArgValue(templateDefinition);
 
-		builder.addConstructorArgValue(element.getAttribute("command"));
-		builder.addConstructorArgValue(element.getAttribute(EXPRESSION_ATTRIBUTE));
+		if (element.hasAttribute("session-callback")) {
+			builder.addConstructorArgReference(element.getAttribute("session-callback"));
+		}
+		else {
+			builder.addConstructorArgValue(element.getAttribute("command"));
+			builder.addConstructorArgValue(element.getAttribute(EXPRESSION_ATTRIBUTE));
+		}
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "command-options", "options");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");

spring-integration-file/src/main/java/org/springframework/integration/file/remote/MessageSessionCallback.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.springframework.integration.file.remote;
+
+import java.io.IOException;
+
+import org.springframework.integration.file.remote.session.Session;
+import org.springframework.messaging.Message;
+
+/**
+ * Callback invoked by {@code RemoteFileOperations.executeForMessage()}
+ * - allows multiple operations on a session.
+ *
+ * @author Artem Bilan
+ * @since 4.2
+ */
+public interface MessageSessionCallback<F, T> {
+
+	/**
+	 * Called within the context of a session and requestMessage.
+	 * Perform some operation(s) on the session.
+	 * The caller will take care of closing the session after this method exits.
+	 *
+	 * @param session The session.
+	 * @param requestMessage The message to take in account with session operation(s).
+	 * @return The result of type T.
+	 * @throws IOException Any IOException.
+	 */
+	T doInSession(Session<F> session, Message<?> requestMessage) throws IOException;
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java

@@ -39,6 +39,7 @@ import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.FileListFilter;
 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.SessionCallback;
 import org.springframework.integration.file.remote.session.Session;
@@ -71,7 +72,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	/**
 	 * Enumeration of commands supported by the gateways.
 	 */
-	public static enum Command {
+	public enum Command {
 
 		/**
 		 * List remote files.
@@ -110,7 +111,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 
 		private String command;
 
-		private Command(String command) {
+		Command(String command) {
 			this.command = command;
 		}
 
@@ -126,13 +127,14 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			}
 			throw new IllegalArgumentException("No Command with value '" + cmd + "'");
 		}
+
 	}
 
 	/**
 	 * Enumeration of options supported by various commands.
 	 *
 	 */
-	public static enum Option {
+	public enum Option {
 
 		/**
 		 * Don't return full file information; just the name (ls).
@@ -181,7 +183,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 
 		private String option;
 
-		private Option(String option) {
+		Option(String option) {
 			this.option = option;
 		}
 
@@ -197,10 +199,13 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			}
 			throw new IllegalArgumentException("No option with value '" + opt + "'");
 		}
+
 	}
 
 	private final ExpressionEvaluatingMessageProcessor<String> fileNameProcessor;
 
+	private final MessageSessionCallback<F, ?> messageSessionCallback;
+
 	private volatile ExpressionEvaluatingMessageProcessor<String> renameProcessor =
 			new ExpressionEvaluatingMessageProcessor<String>(
 					new SpelExpressionParser().parseExpression("headers." + FileHeaders.RENAME_TO));
@@ -226,33 +231,75 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 
 	private volatile FileExistsMode fileExistsMode;
 
+	/**
+	 * Construct an instance using the provided session factory and callback for
+	 * performing operations on the session.
+	 * @param sessionFactory the session factory.
+	 * @param messageSessionCallback the callback.
+	 */
+	public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory,
+			MessageSessionCallback<F, ?> messageSessionCallback) {
+		this(new RemoteFileTemplate<F>(sessionFactory), messageSessionCallback);
+	}
+
+	/**
+	 * Construct an instance with the supplied remote file template and callback
+	 * for performing operations on the session.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param messageSessionCallback the callback.
+	 */
+	public AbstractRemoteFileOutboundGateway(RemoteFileTemplate<F> remoteFileTemplate,
+			MessageSessionCallback<F, ?> messageSessionCallback) {
+		Assert.notNull(remoteFileTemplate, "'remoteFileTemplate' cannot be null");
+		Assert.notNull(messageSessionCallback, "'messageSessionCallback' cannot be null");
+		this.remoteFileTemplate = remoteFileTemplate;
+		this.messageSessionCallback = messageSessionCallback;
+		this.fileNameProcessor = null;
+		this.command = null;
+	}
+
+	/**
+	 * Construct an instance with the supplied session factory, a command ('ls', 'get'
+	 * etc), and an expression to determine the filename.
+	 * @param sessionFactory the session factory.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
 	public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory, String command,
 			String expression) {
-		Assert.notNull(sessionFactory, "'sessionFactory' cannot be null");
-		this.remoteFileTemplate = new RemoteFileTemplate<F>(sessionFactory);
-		this.command = Command.toCommand(command);
-		this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor<String>(
-			new SpelExpressionParser().parseExpression(expression));
+		this(sessionFactory, Command.toCommand(command), expression);
 	}
 
-	public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory, Command command,
-			String expression) {
-		Assert.notNull(sessionFactory, "'sessionFactory' cannot be null");
-		this.remoteFileTemplate = new RemoteFileTemplate<F>(sessionFactory);
-		this.command = command;
-		this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor<String>(
-			new SpelExpressionParser().parseExpression(expression));
+	/**
+	 * Construct an instance with the supplied session factory, a command ('ls', 'get'
+	 * etc), and an expression to determine the filename.
+	 * @param sessionFactory the session factory.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
+	public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory, Command command, String expression) {
+		this(new RemoteFileTemplate<F>(sessionFactory), command, expression);
 	}
 
+	/**
+	 * Construct an instance with the supplied remote file template, a command ('ls',
+	 * 'get' etc), and an expression to determine the filename.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
 	public AbstractRemoteFileOutboundGateway(RemoteFileTemplate<F> remoteFileTemplate, String command,
 			String expression) {
-		Assert.notNull(remoteFileTemplate, "'remoteFileTemplate' cannot be null");
-		this.remoteFileTemplate = remoteFileTemplate;
-		this.command = Command.toCommand(command);
-		this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor<String>(
-			new SpelExpressionParser().parseExpression(expression));
+		this(remoteFileTemplate, Command.toCommand(command), expression);
 	}
 
+	/**
+	 * Construct an instance with the supplied remote file template, a command ('ls',
+	 * 'get' etc), and an expression to determine the filename.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
 	public AbstractRemoteFileOutboundGateway(RemoteFileTemplate<F> remoteFileTemplate, Command command,
 			String expression) {
 		Assert.notNull(remoteFileTemplate, "'remoteFileTemplate' cannot be null");
@@ -260,6 +307,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		this.command = command;
 		this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor<String>(
 			new SpelExpressionParser().parseExpression(expression));
+		this.messageSessionCallback = null;
 	}
 
 	/**
@@ -366,7 +414,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 
 	@Override
 	protected void doInit() {
-		Assert.notNull(this.command, "command must not be null");
+		Assert.state(this.command != null || this.messageSessionCallback != null,
+				"'command' or 'messageSessionCallback' must be specified.");
 		if (Command.RM.equals(this.command) ||
 				Command.GET.equals(this.command)) {
 			Assert.isNull(this.filter, "Filters are not supported with the rm and get commands");
@@ -402,10 +451,10 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		}
 		if (Command.MGET.equals(this.command)) {
 			Assert.isTrue(!(this.options.contains(Option.SUBDIRS)),
-					"Cannot use " + Option.SUBDIRS.toString() + " when using 'mget' use " + Option.RECURSIVE.toString() +
-							" to obtain files in subdirectories");
+					"Cannot use " + Option.SUBDIRS.toString() + " when using 'mget' use "
+							+ Option.RECURSIVE.toString() +	" to obtain files in subdirectories");
 		}
-		if (this.getBeanFactory() != null) {
+		if (this.fileNameProcessor != null && getBeanFactory() != null) {
 			this.fileNameProcessor.setBeanFactory(this.getBeanFactory());
 			this.renameProcessor.setBeanFactory(this.getBeanFactory());
 			this.remoteFileTemplate.setBeanFactory(this.getBeanFactory());
@@ -413,25 +462,33 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	}
 
 	@Override
-	protected Object handleRequestMessage(Message<?> requestMessage) {
-		switch (this.command) {
-		case LS:
-			return doLs(requestMessage);
-		case GET:
-			return doGet(requestMessage);
-		case MGET:
-			return doMget(requestMessage);
-		case RM:
-			return doRm(requestMessage);
-		case MV:
-			return doMv(requestMessage);
-		case PUT:
-			return doPut(requestMessage);
-		case MPUT:
-			return doMput(requestMessage);
-		default:
-			return null;
+	protected Object handleRequestMessage(final Message<?> requestMessage) {
+		if (this.command != null) {
+			switch (this.command) {
+			case LS:
+				return doLs(requestMessage);
+			case GET:
+				return doGet(requestMessage);
+			case MGET:
+				return doMget(requestMessage);
+			case RM:
+				return doRm(requestMessage);
+			case MV:
+				return doMv(requestMessage);
+			case PUT:
+				return doPut(requestMessage);
+			case MPUT:
+				return doMput(requestMessage);
+			}
 		}
+		return this.remoteFileTemplate.execute(new SessionCallback<F, Object>() {
+
+			@Override
+			public Object doInSession(Session<F> session) throws IOException {
+				return messageSessionCallback.doInSession(session, requestMessage);
+			}
+
+		});
 	}
 
 	private Object doLs(Message<?> requestMessage) {

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java

@@ -23,6 +23,7 @@ import java.util.List;
 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;
@@ -32,15 +33,51 @@ import org.springframework.integration.ftp.session.FtpFileInfo;
  * Outbound Gateway for performing remote file operations via FTP/FTPS.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.1
  */
 public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFile> {
 
-	public FtpOutboundGateway(SessionFactory<FTPFile> sessionFactory, String command,
-			String expression) {
+	/**
+	 * Construct an instance using the provided session factory and callback for
+	 * performing operations on the session.
+	 * @param sessionFactory the session factory.
+	 * @param messageSessionCallback the callback.
+	 */
+	public FtpOutboundGateway(SessionFactory<FTPFile> sessionFactory,
+			MessageSessionCallback<FTPFile, ?> messageSessionCallback) {
+		super(sessionFactory, messageSessionCallback);
+	}
+
+	/**
+	 * Construct an instance with the supplied remote file template and callback
+	 * for performing operations on the session.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param messageSessionCallback the callback.
+	 */
+	public FtpOutboundGateway(RemoteFileTemplate<FTPFile> remoteFileTemplate,
+			MessageSessionCallback<FTPFile, ?> messageSessionCallback) {
+		super(remoteFileTemplate, messageSessionCallback);
+	}
+
+	/**
+	 * Construct an instance with the supplied session factory, a command ('ls', 'get'
+	 * etc), and an expression to determine the filename.
+	 * @param sessionFactory the session factory.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
+	public FtpOutboundGateway(SessionFactory<FTPFile> sessionFactory, String command, String expression) {
 		super(sessionFactory, command, expression);
 	}
 
+	/**
+	 * Construct an instance with the supplied remote file template, a command ('ls',
+	 * 'get' etc), and an expression to determine the filename.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
 	public FtpOutboundGateway(RemoteFileTemplate<FTPFile> remoteFileTemplate, String command, String expression) {
 		super(remoteFileTemplate, command, expression);
 	}
@@ -82,7 +119,6 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
 			canonicalFiles.add(new FtpFileInfo(file));
 		}
 		return canonicalFiles;
-
 	}
 
 	@Override
@@ -91,5 +127,4 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
 		return file;
 	}
 
-
 }

spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.2.xsd

@@ -246,7 +246,7 @@
 						<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
 									 minOccurs="0" maxOccurs="1" />
 					</xsd:all>
-					<xsd:attribute name="command" use="required">
+					<xsd:attribute name="command">
 						<xsd:annotation>
 							<xsd:documentation>
 								FTP command.
@@ -256,6 +256,20 @@
 							<xsd:union memberTypes="int-file:remoteGatewayCommand xsd:string"/>
 						</xsd:simpleType>
 					</xsd:attribute>
+					<xsd:attribute name="session-callback" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+											type="org.springframework.integration.file.remote.MessageSessionCallback" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								The 'MessageSessionCallback' bean reference to perform custom operation(s) on 'Session'
+								with 'requestMessage'.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="command-options" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
@@ -272,7 +286,7 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="expression" use="required" type="xsd:string">
+					<xsd:attribute name="expression" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
 								SpEL expression representing the path in the
@@ -462,8 +476,7 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="auto-create-local-directory"
-						type="xsd:boolean">
+					<xsd:attribute name="auto-create-local-directory" type="xsd:boolean">
 						<xsd:annotation>
 							<xsd:documentation>
 								Tells this adapter if local directory must be

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml

@@ -160,4 +160,14 @@
 	<bean id="sortingFilter"
 		class="org.springframework.integration.ftp.outbound.FtpServerOutboundTests$SortingFileListFilter"/>
 
+	<int-ftp:outbound-gateway
+			session-factory="ftpSessionFactory"
+			session-callback="messageSessionCallback"
+			request-channel="inboundCallback"
+			reply-channel="output"/>
+
+	<bean id="messageSessionCallback"
+		  class="org.springframework.integration.ftp.outbound.FtpServerOutboundTests$TestMessageSessionCallback"/>
+
+
 </beans>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java

@@ -60,6 +60,7 @@ import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.InputStreamCallback;
+import org.springframework.integration.file.remote.MessageSessionCallback;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.SessionCallback;
 import org.springframework.integration.file.remote.session.Session;
@@ -135,6 +136,9 @@ public class FtpServerOutboundTests {
 	@Autowired
 	private DirectChannel inboundGetStream;
 
+	@Autowired
+	private DirectChannel inboundCallback;
+
 	@Before
 	public void setup() {
 		this.ftpServer.recursiveDelete(ftpServer.getTargetLocalDirectory());
@@ -535,6 +539,14 @@ public class FtpServerOutboundTests {
 		assertEquals(6, files[0].getSize());
 	}
 
+	@Test
+	public void testMessageSessionCallback() {
+		this.inboundCallback.send(new GenericMessage<String>("foo"));
+		Message<?> receive = this.output.receive(10000);
+		assertNotNull(receive);
+		assertEquals("FOO", receive.getPayload());
+	}
+
 	public static class SortingFileListFilter implements FileListFilter<File> {
 
 		@Override
@@ -560,4 +572,15 @@ public class FtpServerOutboundTests {
 		}
 
 	}
+
+	private static final class TestMessageSessionCallback
+			implements MessageSessionCallback<FTPFile, Object> {
+
+		@Override
+		public Object doInSession(Session<FTPFile> session, Message<?> requestMessage) throws IOException {
+			return ((String) requestMessage.getPayload()).toUpperCase();
+		}
+
+	}
+
 }

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java

@@ -22,6 +22,7 @@ import java.util.List;
 
 import org.springframework.beans.DirectFieldAccessor;
 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;
@@ -33,19 +34,55 @@ import com.jcraft.jsch.ChannelSftp.LsEntry;
  * Outbound Gateway for performing remote file operations via SFTP.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.1
  */
 public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway<LsEntry> {
 
+	/**
+	 * Construct an instance using the provided session factory and callback for
+	 * performing operations on the session.
+	 * @param sessionFactory the session factory.
+	 * @param messageSessionCallback the callback.
+	 */
+	public SftpOutboundGateway(SessionFactory<LsEntry> sessionFactory,
+			MessageSessionCallback<LsEntry, ?> messageSessionCallback) {
+		super(sessionFactory, messageSessionCallback);
+	}
+
+	/**
+	 * Construct an instance with the supplied remote file template and callback
+	 * for performing operations on the session.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param messageSessionCallback the callback.
+	 */
+	public SftpOutboundGateway(RemoteFileTemplate<LsEntry> remoteFileTemplate,
+				MessageSessionCallback<LsEntry, ?> messageSessionCallback) {
+		super(remoteFileTemplate, messageSessionCallback);
+	}
+
+	/**
+	 * Construct an instance with the supplied session factory, a command ('ls', 'get'
+	 * etc), and an expression to determine the filename.
+	 * @param sessionFactory the session factory.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
 	public SftpOutboundGateway(SessionFactory<LsEntry> sessionFactory, String command, String expression) {
 		super(sessionFactory, command, expression);
 	}
 
+	/**
+	 * Construct an instance with the supplied remote file template, a command ('ls',
+	 * 'get' etc), and an expression to determine the filename.
+	 * @param remoteFileTemplate the remote file template.
+	 * @param command the command.
+	 * @param expression the filename expression.
+	 */
 	public SftpOutboundGateway(RemoteFileTemplate<LsEntry> remoteFileTemplate, String command, String expression) {
 		super(remoteFileTemplate, command, expression);
 	}
 
-
 	@Override
 	protected boolean isDirectory(LsEntry file) {
 		return file.getAttrs().isDir();

spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.2.xsd

@@ -250,7 +250,7 @@
 						<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
 									 minOccurs="0" maxOccurs="1" />
 					</xsd:all>
-					<xsd:attribute name="command" use="required">
+					<xsd:attribute name="command">
 						<xsd:annotation>
 							<xsd:documentation>
 								sftp command.
@@ -260,6 +260,20 @@
 							<xsd:union memberTypes="int-file:remoteGatewayCommand xsd:string"/>
 						</xsd:simpleType>
 					</xsd:attribute>
+					<xsd:attribute name="session-callback" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+											type="org.springframework.integration.file.remote.MessageSessionCallback" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								The 'MessageSessionCallback' bean reference to perform custom operation(s) on 'Session'
+								with 'requestMessage'.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="command-options" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
@@ -276,7 +290,7 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="expression" use="required" type="xsd:string">
+					<xsd:attribute name="expression" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
 								SpEL expression representing the path in the

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml

@@ -153,4 +153,13 @@
 	<int:service-activator input-channel="markers"
 	  expression="payload.mark.toString().equals('END') ? headers['file_remoteSession'].close() : null"/>
 
+	<int-sftp:outbound-gateway
+			session-factory="sftpSessionFactory"
+			session-callback="messageSessionCallback"
+			request-channel="inboundCallback"
+			reply-channel="output"/>
+
+	<bean id="messageSessionCallback"
+		  class="org.springframework.integration.sftp.outbound.SftpServerOutboundTests$TestMessageSessionCallback"/>
+
 </beans>

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java

@@ -47,6 +47,7 @@ import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.remote.MessageSessionCallback;
 import org.springframework.integration.file.remote.SessionCallback;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.sftp.TestSftpServer;
@@ -62,6 +63,7 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.FileCopyUtils;
 
+import com.jcraft.jsch.ChannelSftp;
 import com.jcraft.jsch.ChannelSftp.LsEntry;
 
 /**
@@ -129,6 +131,9 @@ public class SftpServerOutboundTests {
 	@Autowired
 	private DirectChannel inboundGetStream;
 
+	@Autowired
+	private DirectChannel inboundCallback;
+
 	@Before
 	@After
 	public void setup() {
@@ -420,6 +425,14 @@ public class SftpServerOutboundTests {
 		assertFalse(((Session<?>) result.getHeaders().get(FileHeaders.REMOTE_SESSION)).isOpen());
 	}
 
+	@Test
+	public void testMessageSessionCallback() {
+		this.inboundCallback.send(new GenericMessage<String>("foo"));
+		Message<?> receive = this.output.receive(10000);
+		assertNotNull(receive);
+		assertEquals("FOO", receive.getPayload());
+	}
+
 	private void assertLength6(SftpRemoteFileTemplate template) {
 		LsEntry[] files = template.execute(new SessionCallback<LsEntry, LsEntry[]>() {
 
@@ -432,4 +445,14 @@ public class SftpServerOutboundTests {
 		assertEquals(6, files[0].getAttrs().getSize());
 	}
 
+	private static final class TestMessageSessionCallback
+			implements MessageSessionCallback<LsEntry, Object> {
+
+		@Override
+		public Object doInSession(Session<ChannelSftp.LsEntry> session, Message<?> requestMessage) throws IOException {
+			return ((String) requestMessage.getPayload()).toUpperCase();
+		}
+
+	}
+
 }

src/reference/asciidoc/ftp.adoc

@@ -581,7 +581,8 @@ If you want your Sessions to be cached, simply configure your default Session Fa
 </bean>
 ----
 
-In the above example you see a `CachingSessionFactory` created with the `sessionCacheSize` set to 10 and the `sessionWaitTimeout` set to 1 second (its value is in millliseconds).
+In the above example you see a `CachingSessionFactory` created with the `sessionCacheSize` set to 10 and the
+`sessionWaitTimeout` set to 1 second (its value is in milliseconds).
 
 Starting with _Spring Integration version 3.0_, the `CachingConnectionFactory` provides a `resetCache()` method.
 When invoked, all idle sessions are immediately closed and in-use sessions are closed when they are returned to the cache.
@@ -594,6 +595,35 @@ Starting with _Spring Integration version 3.0_ a new abstraction is provided ove
 The template provides methods to send, retrieve (as an `InputStream`), remove, and rename files.
 In addition an `execute` method is provided allowing the caller to execute multiple operations on the session.
 In all cases, the template takes care of reliably closing the session.
-For more information, refer to the http://docs.spring.io/spring-integration/api/org/springframework/integration/file/remote/RemoteFileTemplate.html[javadocs for `RemoteFileTemplate`] There is a subclass for FTP: `FtpRemoteFileTemplate`.
+For more information, refer to the
+http://docs.spring.io/spring-integration/api/org/springframework/integration/file/remote/RemoteFileTemplate.html[JavaDocs for `RemoteFileTemplate`].
+There is a subclass for FTP: `FtpRemoteFileTemplate`.
 
 Additional methods were added in _version 4.1_ including `getClientInstance()` which provides access to the underlying `FTPClient` enabling access to low-level APIs.
+
+[[ftp-session-callback]]
+=== MessageSessionCallback
+
+Starting with _Spring Integration version 4.2_, a `MessageSessionCallback<F, T>` implementation can be used with the
+`<int-ftp:outbound-gateway/>` (`FtpOutboundGateway`) to perform any operation(s) on the `Session<FTPFile>` with
+the `requestMessage` context.
+It can be used for any non-standard or low-level FTP operation (or several); for example, allowing access
+from an integration flow definition, and _functional_ interface (Lambda) implementation injection:
+
+[source,java]
+----
+@Bean
+@ServiceActivator(inputChannel = "ftpChannel")
+public MessageHandler ftpOutboundGateway(SessionFactory<FTPFile> sessionFactory) {
+    return new FtpOutboundGateway(sessionFactory,
+         (session, requestMessage) -> session.list(requestMessage.getPayload()));
+}
+----
+
+Another example might be to pre- or post- process the file data being sent/retrieved.
+
+When using XML configuration, the `<int-ftp:outbound-gateway/>` provides a `session-callback` attribute to allow you to
+specify the `MessageSessionCallback` bean name.
+
+NOTE: The `session-callback` is mutually exclusive with the `command` and `expression` attributes.
+When configuring with Java, different constructors are available in the `FtpOutboundGateway` class.

src/reference/asciidoc/sftp.adoc

@@ -244,7 +244,7 @@ If you want your Sessions to be cached, simply configure your default Session Fa
 </bean>
 ----
 
-In the above example you see a `CachingSessionFactory` created with the `sessionCacheSize` set to 10 and the `sessionWaitTimeout` set to 1 second (its value is in millliseconds).
+In the above example you see a `CachingSessionFactory` created with the `sessionCacheSize` set to 10 and the `sessionWaitTimeout` set to 1 second (its value is in milliseconds).
 
 Starting with _Spring Integration version 3.0_, the `CachingConnectionFactory` provides a `resetCache()` method.
 When invoked, all idle sessions are immediately closed and in-use sessions are closed when they are returned to the cache.
@@ -646,3 +646,30 @@ For example, here is valid configuration of a logger using Log4J.
 ----
 log4j.category.com.jcraft.jsch=DEBUG
 ----
+
+[[sftp-session-callback]]
+=== MessageSessionCallback
+
+Starting with _Spring Integration version 4.2_, a `MessageSessionCallback<F, T>` implementation can be used with the
+`<int-ftp:outbound-gateway/>` (`FtpOutboundGateway`) to perform any operation(s) on the `Session<FTPFile>` with
+the `requestMessage` context.
+It can be used for any non-standard or low-level FTP operation (or several); for example, allowing access
+from an integration flow definition, and _functional_ interface (Lambda) implementation injection:
+
+[source,java]
+----
+@Bean
+@ServiceActivator(inputChannel = "sftpChannel")
+public MessageHandler sftpOutboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+    return new SftpOutboundGateway(sessionFactory,
+         (session, requestMessage) -> session.list(requestMessage.getPayload()));
+}
+----
+
+Another example might be to pre- or post- process the file data being sent/retrieved.
+
+When using XML configuration, the `<int-sftp:outbound-gateway/>` provides a `session-callback` attribute to allow you
+to specify the `MessageSessionCallback` bean name.
+
+NOTE: The `session-callback` is mutually exclusive with the `command` and `expression` attributes.
+When configuring with Java, different constructors are available in the `SftpOutboundGateway` class.

src/reference/asciidoc/whats-new.adoc

@@ -323,6 +323,14 @@ false).
 
 See <<sftp-unk-hosts>> for more information.
 
+===== MessageSessionCallback
+
+The `MessageSessionCallback<F, T>` has been introduced to perform any custom `Session` operation(s) with the
+`requestMessage` context in the `<int-(s)ftp:outbound-gateway/>`.
+
+See <<ftp-session-callback>> and <<sftp-session-callback>> for more information.
+
+
 ==== Websocket Changes
 
 `WebSocketHandlerDecoratorFactory` support has been added to the `ServerWebSocketContainer`