INT-2633 Add File Disposition to (S)FTP Inbound

FileReadingMessageSource supports disposition of the payload
after the message is sent (or via transaction synchronization).

The (S)FTP adapters delegate to an FRMS; add support for the
file disposition expression, result channel, and send timeout
to the (S)FTP adapters.

Also tested with ftp sample - INTSAMPLES-83 - patch will be
committed once this is in a milestone.

spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-2012 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;
+
+import java.io.File;
+
+import org.springframework.integration.Message;
+
+/**
+ * A simple wrapper for a Message<File>; used for
+ * file disposition after the send completes, or
+ * after the transaction commits with a transactional
+ * poller.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class FileMessageHolder {
+
+	private Message<File> message;
+
+	Message<File> getMessage() {
+		return message;
+	}
+
+	void setMessage(Message<File> message) {
+		this.message = message;
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java

@@ -38,7 +38,6 @@ import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.core.PseudoTransactionalMessageSource;
-import org.springframework.integration.file.FileReadingMessageSource.FileMessageHolder;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.support.MessageBuilder;
@@ -389,16 +388,4 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 		this.afterCommit(resource);
 	}
 
-	class FileMessageHolder {
-
-		private Message<File> message;
-
-		Message<File> getMessage() {
-			return message;
-		}
-
-		void setMessage(Message<File> message) {
-			this.message = message;
-		}
-	}
 }

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

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.file.config;
 
-import org.w3c.dom.Element;
-
 import org.springframework.beans.BeanMetadataElement;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
@@ -26,10 +24,11 @@ import org.springframework.integration.config.xml.AbstractPollingInboundChannelA
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.file.remote.session.SessionFactoryFactoryBean;
 import org.springframework.util.StringUtils;
+import org.w3c.dom.Element;
 
 /**
  * Abstract base class for parsing remote file inbound channel adapters.
- * 
+ *
  * @author Oleg Zhurakousky
  * @author Mark Fisher
  * @since 2.0
@@ -44,9 +43,9 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 		BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(SessionFactoryFactoryBean.class);
 		sessionFactoryBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
 		sessionFactoryBuilder.addConstructorArgValue(element.getAttribute("cache-sessions"));
-		
+
 		synchronizerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
-		
+
 		// configure the InboundFileSynchronizer properties
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "delete-remote-files");
@@ -71,6 +70,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 			localFileGeneratorExpressionBuilder.addConstructorArgValue(localFileGeneratorExpression);
 			synchronizerBuilder.addPropertyValue("localFilenameGeneratorExpression", localFileGeneratorExpressionBuilder.getBeanDefinition());
 		}
+		FileNamespaceUtils.setDispositionAttributes(element, messageSourceBuilder);
 		return messageSourceBuilder.getBeanDefinition();
 	}
 

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

@@ -21,9 +21,7 @@ import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
-import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.file.locking.NioFileLocker;
@@ -49,14 +47,7 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size");
-		String dispositionExpression = element.getAttribute("disposition-expression");
-		if (StringUtils.hasText(dispositionExpression)) {
-			RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
-			expressionDef.getConstructorArgumentValues().addGenericArgumentValue(dispositionExpression);
-			builder.addPropertyValue("dispositionExpression", expressionDef);
-		}
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "disposition-result-channel");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "disposition-send-timeout");
+		FileNamespaceUtils.setDispositionAttributes(element, builder);
 		String filterBeanName = this.registerFilter(element, parserContext);
 		String lockerBeanName = registerLocker(element, parserContext);
 		if (lockerBeanName != null) {

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.file.config;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.integration.config.ExpressionFactoryBean;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.util.StringUtils;
+import org.w3c.dom.Element;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class FileNamespaceUtils {
+
+	public static void setDispositionAttributes(Element element, BeanDefinitionBuilder builder) {
+		String dispositionExpression = element.getAttribute("disposition-expression");
+		if (StringUtils.hasText(dispositionExpression)) {
+			RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
+			expressionDef.getConstructorArgumentValues().addGenericArgumentValue(dispositionExpression);
+			builder.addPropertyValue("dispositionExpression", expressionDef);
+		}
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "disposition-result-channel");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "disposition-send-timeout");
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -22,10 +22,13 @@ import java.util.Arrays;
 import java.util.Comparator;
 import java.util.regex.Pattern;
 
+import org.springframework.expression.Expression;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessagingException;
-import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.core.PseudoTransactionalMessageSource;
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.file.FileMessageHolder;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
@@ -50,11 +53,13 @@ import org.springframework.util.Assert;
  * {@link AbstractInboundFileSynchronizer}. The synchronizer must
  * handle the work of actually connecting to the remote file system and
  * delivering new {@link File}s.
- * 
+ *
  * @author Josh Long
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
-public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends MessageProducerSupport implements MessageSource<File> {
+public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends MessageProducerSupport
+	implements PseudoTransactionalMessageSource<File, FileMessageHolder> {
 
 	/**
 	 * Should the endpoint attempt to create the local directory? True by default.
@@ -81,7 +86,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
 	public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer) {
 		this(synchronizer, null);
 	}
-	
+
 	public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer, Comparator<File> comparator) {
 		Assert.notNull(synchronizer, "synchronizer must not be null");
 		this.synchronizer = synchronizer;
@@ -90,7 +95,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
 		}
 		else {
 			this.fileSource = new FileReadingMessageSource(comparator);
-		}	
+		}
 	}
 
 
@@ -102,6 +107,18 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
 		this.localDirectory = localDirectory;
 	}
 
+	public void setDispositionExpression(Expression dispositionExpression) {
+		this.fileSource.setDispositionExpression(dispositionExpression);
+	}
+
+	public void setDispositionResultChannel(MessageChannel dispositionResultChannel) {
+		this.fileSource.setDispositionResultChannel(dispositionResultChannel);
+	}
+
+	public void setDispositionSendTimeout(long dispositionSendTimeout) {
+		this.fileSource.setDispositionSendTimeout(dispositionSendTimeout);
+	}
+
 	@Override
 	protected void onInit() {
 		Assert.notNull(this.localDirectory, "localDirectory must not be null");
@@ -155,4 +172,24 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
 				new RegexPatternFileListFilter(completePattern)));
 	}
 
+	public FileMessageHolder getResource() {
+		return this.fileSource.getResource();
+	}
+
+	public void afterCommit(FileMessageHolder resource) {
+		this.fileSource.afterCommit(resource);
+	}
+
+	public void afterRollback(FileMessageHolder resource) {
+		this.fileSource.afterRollback(resource);
+	}
+
+	public void afterReceiveNoTx(FileMessageHolder resource) {
+		this.fileSource.afterReceiveNoTx(resource);
+	}
+
+	public void afterSendNoTx(FileMessageHolder resource) {
+		this.fileSource.afterSendNoTx(resource);
+	}
+
 }

spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java

@@ -40,7 +40,6 @@ import org.mockito.runners.MockitoJUnitRunner;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.Message;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.file.FileReadingMessageSource.FileMessageHolder;
 import org.springframework.integration.message.GenericMessage;
 
 /**

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -17,36 +17,38 @@
 package org.springframework.integration.ftp.config;
 
 import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser;
+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;
 
 /**
  * Parser for the FTP 'inbound-channel-adapter' element.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
 public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundChannelAdapterParser {
 
-	private static final String BASE_PACKAGE = "org.springframework.integration.ftp";
-
-
 	@Override
 	protected String getMessageSourceClassname() {
-		return BASE_PACKAGE + ".inbound.FtpInboundFileSynchronizingMessageSource";
+		return FtpInboundFileSynchronizingMessageSource.class.getName();
 	}
 
 	@Override
 	protected String getInboundFileSynchronizerClassname() {
-		return BASE_PACKAGE + ".inbound.FtpInboundFileSynchronizer";
+		return FtpInboundFileSynchronizer.class.getName();
 	}
 
 	@Override
 	protected String getSimplePatternFileListFilterClassname() {
-		return BASE_PACKAGE + ".filters.FtpSimplePatternFileListFilter";
+		return FtpSimplePatternFileListFilter.class.getName();
 	}
 
 	@Override
 	protected String getRegexPatternFileListFilterClassname() {
-		return BASE_PACKAGE + ".filters.FtpRegexPatternFileListFilter";
+		return FtpRegexPatternFileListFilter.class.getName();
 	}
 
 }

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

@@ -218,6 +218,42 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="disposition-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								SpEL expression to be executed after the message has been sent. If running in a transactional
+								poller, it will be executed after the transaction commits. If running in a non-transactional
+								poller it will execute after the message is sent. Note that the actual point of execution
+								depends on any asynchronous handoffs on the downstream flow. It will be executed when the
+								current thread returns from the channel send. The root object of the expression is the
+								original message (with a File payload). Examples: "payload.delete()",
+								"payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)".
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="disposition-result-channel" type="xsd:string" default="nullChannel">
+						<xsd:annotation>
+							<xsd:documentation>
+								If a 'disposition-expression' is provided, and that expression returns a result, the result
+								is sent to this channel, with the original message payload and a 'file_dispositionResult'
+								header containing the result of the expression execution.
+								<xsd:appinfo>
+									<tool:annotation kind="ref">
+										<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+									</tool:annotation>
+								</xsd:appinfo>
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="disposition-send-timeout" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								If a 'disposition-expression' is provided, and that expression returns a result, the result
+								is sent to this disposition-result-channel. This timeout specifies how long to wait if
+								that channel might block (such as a bounded queue channel that is full). Default infinity.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml

@@ -24,9 +24,14 @@
 				local-filename-generator-expression="#this.toUpperCase() + '.a'"
 				comparator="comparator"
 				temporary-file-suffix=".foo"
-				remote-directory="foo/bar">
+				remote-directory="foo/bar"
+				disposition-expression="'foo'"
+				disposition-result-channel="dispoChannel"
+				disposition-send-timeout="123">
 			<int:poller fixed-rate="1000"/>
 	</int-ftp:inbound-channel-adapter>
+
+	<int:channel id="dispoChannel" />
 	
 	<bean id="comparator" class="org.mockito.Mockito" factory-method="mock">
 		<constructor-arg value="java.util.Comparator"/>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java

@@ -18,7 +18,6 @@ package org.springframework.integration.ftp.config;
 
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertNotNull;
-
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
@@ -30,12 +29,13 @@ import java.util.Comparator;
 import java.util.Map;
 
 import org.junit.Test;
-
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.expression.Expression;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
@@ -54,7 +54,7 @@ public class FtpInboundChannelAdapterParserTests {
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testFtpInboundChannelAdapterComplete() throws Exception{
-		ApplicationContext ac = 
+		ApplicationContext ac =
 			new ClassPathXmlApplicationContext("FtpInboundChannelAdapterParserTests-context.xml", this.getClass());
 		SourcePollingChannelAdapter adapter = ac.getBean("ftpInbound", SourcePollingChannelAdapter.class);
 		assertFalse(TestUtils.getPropertyValue(adapter, "autoStartup", Boolean.class));
@@ -64,10 +64,10 @@ public class FtpInboundChannelAdapterParserTests {
 		assertEquals("ftp:inbound-channel-adapter", adapter.getComponentType());
 		assertNotNull(TestUtils.getPropertyValue(adapter, "poller"));
 		assertEquals(ac.getBean("ftpChannel"), TestUtils.getPropertyValue(adapter, "outputChannel"));
-		FtpInboundFileSynchronizingMessageSource inbound = 
+		FtpInboundFileSynchronizingMessageSource inbound =
 			(FtpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(adapter, "source");
-		
-		FtpInboundFileSynchronizer fisync = 
+
+		FtpInboundFileSynchronizer fisync =
 			(FtpInboundFileSynchronizer) TestUtils.getPropertyValue(inbound, "synchronizer");
 		assertNotNull(TestUtils.getPropertyValue(fisync, "localFilenameGeneratorExpression"));
 		assertEquals(".foo", TestUtils.getPropertyValue(fisync, "temporaryFileSuffix", String.class));
@@ -78,6 +78,12 @@ public class FtpInboundChannelAdapterParserTests {
 		assertNotNull(filter);
 		Object sessionFactory = TestUtils.getPropertyValue(fisync, "sessionFactory");
 		assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass()));
+		FileReadingMessageSource source = TestUtils.getPropertyValue(inbound, "fileSource", FileReadingMessageSource.class);
+		assertEquals("foo", TestUtils.getPropertyValue(source, "dispositionExpression", Expression.class).getValue());
+		assertSame(ac.getBean("dispoChannel"), TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(source, "dispositionMessagingTemplate"), "defaultChannel"));
+		assertEquals(123L, TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(source, "dispositionMessagingTemplate"), "sendTimeout"));
 	}
 
 	@Test
@@ -87,7 +93,7 @@ public class FtpInboundChannelAdapterParserTests {
 		SourcePollingChannelAdapter adapter = ac.getBean("simpleAdapter", SourcePollingChannelAdapter.class);
 		Object sessionFactory = TestUtils.getPropertyValue(adapter, "source.synchronizer.sessionFactory");
 		assertEquals(CachingSessionFactory.class, sessionFactory.getClass());
-		FtpInboundFileSynchronizer fisync = 
+		FtpInboundFileSynchronizer fisync =
 			TestUtils.getPropertyValue(adapter, "source.synchronizer", FtpInboundFileSynchronizer.class);
 		String remoteFileSeparator = (String) TestUtils.getPropertyValue(fisync, "remoteFileSeparator");
 		assertNotNull(remoteFileSeparator);
@@ -96,7 +102,7 @@ public class FtpInboundChannelAdapterParserTests {
 
 	@Test
 	public void testFtpInboundChannelAdapterCompleteNoId() throws Exception{
-		ApplicationContext ac = 
+		ApplicationContext ac =
 			new ClassPathXmlApplicationContext("FtpInboundChannelAdapterParserTests-context.xml", this.getClass());
 		Map<String, SourcePollingChannelAdapter> spcas = ac.getBeansOfType(SourcePollingChannelAdapter.class);
 		SourcePollingChannelAdapter adapter = null;

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -17,36 +17,38 @@
 package org.springframework.integration.sftp.config;
 
 import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser;
+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;
 
 /**
  * Parser for 'sftp:inbound-channel-adapter'
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
 public class SftpInboundChannelAdapterParser extends AbstractRemoteFileInboundChannelAdapterParser {
 
-	private static final String BASE_PACKAGE = "org.springframework.integration.sftp";
-
-
 	@Override
 	protected String getMessageSourceClassname() {
-		return BASE_PACKAGE + ".inbound.SftpInboundFileSynchronizingMessageSource";
+		return SftpInboundFileSynchronizingMessageSource.class.getName();
 	}
 
 	@Override
 	protected String getInboundFileSynchronizerClassname() {
-		return BASE_PACKAGE + ".inbound.SftpInboundFileSynchronizer";
+		return SftpInboundFileSynchronizer.class.getName();
 	}
 
 	@Override
 	protected String getSimplePatternFileListFilterClassname() {
-		return BASE_PACKAGE + ".filters.SftpSimplePatternFileListFilter";
+		return SftpSimplePatternFileListFilter.class.getName();
 	}
 
 	@Override
 	protected String getRegexPatternFileListFilterClassname() {
-		return BASE_PACKAGE + ".filters.SftpRegexPatternFileListFilter";
+		return SftpRegexPatternFileListFilter.class.getName();
 	}
 
 }

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

@@ -221,6 +221,42 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="disposition-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								SpEL expression to be executed after the message has been sent. If running in a transactional
+								poller, it will be executed after the transaction commits. If running in a non-transactional
+								poller it will execute after the message is sent. Note that the actual point of execution
+								depends on any asynchronous handoffs on the downstream flow. It will be executed when the
+								current thread returns from the channel send. The root object of the expression is the
+								original message (with a File payload). Examples: "payload.delete()",
+								"payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)".
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="disposition-result-channel" type="xsd:string" default="nullChannel">
+						<xsd:annotation>
+							<xsd:documentation>
+								If a 'disposition-expression' is provided, and that expression returns a result, the result
+								is sent to this channel, with the original message payload and a 'file_dispositionResult'
+								header containing the result of the expression execution.
+								<xsd:appinfo>
+									<tool:annotation kind="ref">
+										<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+									</tool:annotation>
+								</xsd:appinfo>
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="disposition-send-timeout" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								If a 'disposition-expression' is provided, and that expression returns a result, the result
+								is sent to this disposition-result-channel. This timeout specifies how long to wait if
+								that channel might block (such as a bounded queue channel that is full). Default infinity.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml

@@ -48,9 +48,14 @@
 			local-filename-generator-expression="#this.toUpperCase() + '.a'"
 			temporary-file-suffix=".bar"
 			comparator="comparator"
-			delete-remote-files="${delete.remote.files}">
+			delete-remote-files="${delete.remote.files}"
+			disposition-expression="'foo'"
+			disposition-result-channel="dispoChannel"
+			disposition-send-timeout="123">
 		<poller fixed-rate="1000"/>
 	</sftp:inbound-channel-adapter>
+
+	<channel id="dispoChannel" />
 	
 	<beans:bean id="comparator" class="org.mockito.Mockito" factory-method="mock">
 		<beans:constructor-arg value="java.util.Comparator"/>

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java

@@ -29,14 +29,15 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
-
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.expression.Expression;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
 import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource;
 import org.springframework.integration.test.util.TestUtils;
@@ -46,17 +47,17 @@ import org.springframework.integration.test.util.TestUtils;
  * @author Gary Russell
  */
 public class InboundChannelAdapterParserTests {
-	
+
 	@Before
 	public void prepare(){
 		new File("foo").delete();
 	}
-	
+
 	@Test
 	public void testAutoStartup() throws Exception{
 		ApplicationContext context =
 			new ClassPathXmlApplicationContext("SftpInboundAutostartup-context.xml", this.getClass());
-		
+
 		SourcePollingChannelAdapter adapter = context.getBean("sftpAutoStartup", SourcePollingChannelAdapter.class);
 		assertFalse(adapter.isRunning());
 	}
@@ -67,10 +68,10 @@ public class InboundChannelAdapterParserTests {
 		ApplicationContext context =
 			new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass());
 		assertTrue(new File("src/main/resources").exists());
-	
+
 		Object adapter = context.getBean("sftpAdapterAutoCreate");
 		assertTrue(adapter instanceof SourcePollingChannelAdapter);
-		SftpInboundFileSynchronizingMessageSource source = 
+		SftpInboundFileSynchronizingMessageSource source =
 			(SftpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(adapter, "source");
 		assertNotNull(source);
 		Comparator<File> comparator = TestUtils.getPropertyValue(adapter, "source.fileSource.toBeReceived.q.comparator", Comparator.class);
@@ -83,6 +84,12 @@ public class InboundChannelAdapterParserTests {
 		assertEquals(".", remoteFileSeparator);
 		PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class);
 		assertNotNull(requestChannel.receive(2000));
+		FileReadingMessageSource fileSource = TestUtils.getPropertyValue(source, "fileSource", FileReadingMessageSource.class);
+		assertEquals("foo", TestUtils.getPropertyValue(fileSource, "dispositionExpression", Expression.class).getValue());
+		assertSame(context.getBean("dispoChannel"), TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(fileSource, "dispositionMessagingTemplate"), "defaultChannel"));
+		assertEquals(123L, TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(fileSource, "dispositionMessagingTemplate"), "sendTimeout"));
 	}
 
 	@Test
@@ -108,19 +115,19 @@ public class InboundChannelAdapterParserTests {
 		new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass());
 		assertTrue(new File("target").exists());
 	}
-	
+
 	@Test
 	public void testLocalDirAutoCreated() throws Exception{
 		assertFalse(new File("foo").exists());
 		new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass());
 		assertTrue(new File("foo").exists());
 	}
-	
+
 	@Test(expected=BeanCreationException.class)
 	public void testLocalDirAutoCreateFailed() throws Exception{
 		new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context-fail-autocreate.xml", this.getClass());
 	}
-	
+
 	@After
 	public void cleanUp() throws Exception{
 		new File("foo").delete();