INT-1849/INT-2147 Add Disposition to File Adapter

* Add *disposition-expression* to the inbound file adapter. This allows for operations such as *payload.delete()*, *payload.renameTo()* after the file is processed.
* The result of executing the expression (if any) is sent to the disposition-result-channel.

build.gradle

@@ -254,6 +254,7 @@ project('spring-integration-file') {
 			'org.springframework.integration.*;version="[2.2.0, 2.2.1)"',
 			'org.springframework.beans.*;version="[3.1.1, 4.0.0)"',
 			'org.springframework.expression.*;version="[3.1.1, 4.0.0)"',
+			'org.springframework.context.expression.*;version="[3.1.1, 4.0.0)"',
 			'org.springframework.context;version="[3.1.1, 4.0.0)"',
 			'org.springframework.context.expression.*;version="[3.1.1, 4.0.0)"',
 			'org.springframework.core.*;version="[3.1.1, 4.0.0)"',

spring-integration-core/src/main/java/org/springframework/integration/core/PseudoTransactionalMessageSource.java

@@ -38,27 +38,40 @@ import org.springframework.transaction.support.TransactionSynchronization;
  * @since 2.2
  *
  */
-public interface PseudoTransactionalMessageSource<T> extends MessageSource<T> {
+public interface PseudoTransactionalMessageSource<T, V> extends MessageSource<T> {
 
 	/**
 	 * Obtain the resource on which appropriate action needs
 	 * to be taken.
 	 * @return The resource.
 	 */
-	Object getResource();
+	V getResource();
 
 	/**
 	 * Invoked via {@link TransactionSynchronization} when the
 	 * transaction commits.
 	 * @param resource The resource to be "committed"
 	 */
-	void afterCommit(Object resource);
+	void afterCommit(V resource);
 
 	/**
 	 * Invoked via {@link TransactionSynchronization} when the
 	 * transaction rolls back.
 	 * @param resource
 	 */
-	void afterRollback(Object resource);
+	void afterRollback(V resource);
+
+	/**
+	 * Called when there is no transaction and the receive() call completed.
+	 * @param resource
+	 */
+	void afterReceiveNoTx(V resource);
+
+	/**
+	 * Called when there is no transaction and after the message was
+	 * sent to the channel.
+	 * @param resource
+	 */
+	void afterSendNoTx(V resource);
 
 }

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java

@@ -99,13 +99,14 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 		super.onInit();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Override
 	protected boolean doPoll() {
 		boolean isInTx = false;
-		PseudoTransactionalMessageSource<?> messageSource = null;
+		PseudoTransactionalMessageSource<?,Object> messageSource = null;
 		Object resource = null;
 		if (this.isPseudoTxMessageSource) {
-			messageSource = (PseudoTransactionalMessageSource<?>) this.source;
+			messageSource = (PseudoTransactionalMessageSource<?,Object>) this.source;
 			resource = messageSource.getResource();
 			Assert.state(resource != null, "Pseudo Transactional Message Source returned null resource");
 			if (this.synchronizedTx && TransactionSynchronizationManager.isActualTransactionActive()) {
@@ -122,17 +123,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 		}
 		finally {
 			if (this.isPseudoTxMessageSource && !isInTx) {
-				/*
-				 * If the message source implements PseudoTransactionalMessageSource and
-				 * we're running from a transactional poller, the message source's afterCommit
-				 * method will be called by the transaction interceptor, using the transaction
-				 * synchronization callback, after the transaction is committed.
-				 *
-				 * If we are not running in a transaction, we invoke it manually, so the message
-				 * source can take the appropriate action, immediately after the receive;
-				 * this was the behavior before pseudo transaction support was added.
-				 */
-				messageSource.afterCommit(resource);
+				messageSource.afterReceiveNoTx(resource);
 			}
 		}
 		if (this.logger.isDebugEnabled()){
@@ -143,6 +134,9 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 				message = MessageHistory.write(message, this);
 			}
 			this.messagingTemplate.send(this.outputChannel, message);
+			if (this.isPseudoTxMessageSource && !isInTx) {
+				messageSource.afterSendNoTx(resource);
+			}
 			return true;
 		}
 		if (this.logger.isDebugEnabled()){
@@ -191,21 +185,23 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 			return false;
 		}
 
+		@SuppressWarnings("unchecked")
 		@Override
 		protected void processResourceAfterCommit(PseudoTransactionalResourceHolder resourceHolder) {
 			if (logger.isTraceEnabled()) {
 				logger.trace("'Committing' pseudo-transactional resource");
 			}
-			((PseudoTransactionalMessageSource<?>) source).afterCommit(resourceHolder.getResource());
+			((PseudoTransactionalMessageSource<?,Object>) source).afterCommit(resourceHolder.getResource());
 		}
 
+		@SuppressWarnings("unchecked")
 		@Override
 		public void afterCompletion(int status) {
 			if (status != TransactionSynchronization.STATUS_COMMITTED) {
 				if (logger.isTraceEnabled()) {
 					logger.trace("'Rolling back' pseudo-transactional resource");
 				}
-				((PseudoTransactionalMessageSource<?>) source).afterRollback(this.resourceHolder.getResource());
+				((PseudoTransactionalMessageSource<?,Object>) source).afterRollback(this.resourceHolder.getResource());
 			}
 			super.afterCompletion(status);
 		}

spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java

@@ -0,0 +1,86 @@
+/*
+ * 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.util;
+
+import org.springframework.context.expression.MapAccessor;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.expression.BeanResolver;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.expression.spel.support.StandardTypeConverter;
+
+/**
+ * Utility class with static methods for helping with establishing environments for
+ * SpEL expressions.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class ExpressionUtils {
+
+	/**
+	 * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its
+	 * property accessor property.
+	 * @return the evaluation context.
+	 */
+	public static StandardEvaluationContext createStandardEvaluationContext() {
+		return createStandardEvaluationContext(null, null);
+	}
+
+	/**
+	 * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its
+	 * property accessor property and the supplied {@link BeanResolver} in its
+	 * beanResolver property.
+	 * @param beanResolver the bean factory.
+	 * @return the evaluation context.
+	 */
+	public static StandardEvaluationContext createStandardEvaluationContext(BeanResolver beanResolver) {
+		return createStandardEvaluationContext(beanResolver, null);
+	}
+
+	/**
+	 * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its
+	 * property accessor property and the supplied {@link ConversionService} in its
+	 * conversionService property.
+	 * @param conversionService the conversion service.
+	 * @return the evaluation context.
+	 */
+	public static StandardEvaluationContext createStandardEvaluationContext(ConversionService conversionService) {
+		return createStandardEvaluationContext(null, conversionService);
+	}
+
+	/**
+	 * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its
+	 * property accessor property, the supplied {@link BeanResolver} in its
+	 * beanResolver property, and the supplied {@link ConversionService} in its
+	 * conversionService property.
+	 * @param beanResolver the bean factory.
+	 * @param conversionService the conversion service.
+	 * @return the evaluation context.
+	 */
+	public static StandardEvaluationContext createStandardEvaluationContext(BeanResolver beanResolver,
+			ConversionService conversionService) {
+		StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
+		evaluationContext.addPropertyAccessor(new MapAccessor());
+		if (beanResolver != null) {
+			evaluationContext.setBeanResolver(beanResolver);
+		}
+		if (conversionService != null) {
+			evaluationContext.setTypeConverter(new StandardTypeConverter(conversionService));
+		}
+		return evaluationContext;
+	}
+}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java

@@ -44,7 +44,7 @@ public class PseudoTransactionalMessageSourceTests {
 		final Object object = new Object();
 		final AtomicReference<Object> committed = new AtomicReference<Object>();
 		final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
-		adapter.setSource(new PseudoTransactionalMessageSource<String>() {
+		adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
 
 			public Message<String> receive() {
 				return new GenericMessage<String>("foo");
@@ -61,6 +61,12 @@ public class PseudoTransactionalMessageSourceTests {
 			public void afterRollback(Object resource) {
 				rolledBack.set(resource);
 			}
+
+			public void afterReceiveNoTx(Object resource) {
+			}
+
+			public void afterSendNoTx(Object resource) {
+			}
 		});
 
 		TransactionSynchronizationManager.initSynchronization();
@@ -81,7 +87,7 @@ public class PseudoTransactionalMessageSourceTests {
 		final Object object = new Object();
 		final AtomicReference<Object> committed = new AtomicReference<Object>();
 		final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
-		adapter.setSource(new PseudoTransactionalMessageSource<String>() {
+		adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
 
 			public Message<String> receive() {
 				return new GenericMessage<String>("foo");
@@ -98,6 +104,12 @@ public class PseudoTransactionalMessageSourceTests {
 			public void afterRollback(Object resource) {
 				rolledBack.set(resource);
 			}
+
+			public void afterReceiveNoTx(Object resource) {
+			}
+
+			public void afterSendNoTx(Object resource) {
+			}
 		});
 
 		TransactionSynchronizationManager.initSynchronization();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 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.
@@ -19,8 +19,9 @@ package org.springframework.integration.file;
 /**
  * Pre-defined header names to be used when storing or retrieving
  * File-related values to/from integration Message Headers.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public abstract class FileHeaders {
 
@@ -34,4 +35,6 @@ public abstract class FileHeaders {
 
 	public static final String REMOTE_FILE = PREFIX + "remoteFile";
 
+	public static final String DISPOSITION_RESULT = PREFIX + "dispositionResult";
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 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.
@@ -26,14 +26,23 @@ import java.util.concurrent.PriorityBlockingQueue;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.context.expression.BeanFactoryResolver;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
 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;
+import org.springframework.integration.util.ExpressionUtils;
 import org.springframework.util.Assert;
 
 /**
@@ -50,7 +59,7 @@ import org.springframework.util.Assert;
  * file-writing process renames each file as soon as it is ready for reading. A
  * pattern-matching filter that accepts only files that are ready (e.g. based on
  * a known suffix), composed with the default {@link AcceptOnceFileListFilter}
- * would allow for this. 
+ * would allow for this.
  * <p/>
  * A {@link Comparator} can be used to ensure internal ordering of the Files in
  * a {@link PriorityBlockingQueue}. This does not provide the same guarantees as
@@ -59,12 +68,13 @@ import org.springframework.util.Assert;
  * <p/>
  * FileReadingMessageSource is fully thread-safe under concurrent
  * <code>receive()</code> invocations and message delivery callbacks.
- * 
+ *
  * @author Iwein Fuld
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
-public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource<File> {
+public class FileReadingMessageSource extends IntegrationObjectSupport implements PseudoTransactionalMessageSource<File, FileMessageHolder> {
 
 	private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5;
 
@@ -86,6 +96,16 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 
 	private volatile boolean scanEachPoll = false;
 
+	private volatile Expression dispositionExpression;
+
+	private final MessagingTemplate dispositionMessagingTemplate = new MessagingTemplate();
+
+	private volatile boolean dispostionResultChannelSet;
+
+	private final ThreadLocal<FileMessageHolder> resources = new ThreadLocal<FileMessageHolder>();
+
+	private EvaluationContext evaluationContext = new StandardEvaluationContext();
+
 
 	/**
 	 * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity.
@@ -98,7 +118,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	 * Creates a FileReadingMessageSource with a bounded queue of the given
 	 * capacity. This can be used to reduce the memory footprint of this
 	 * component when reading from a large directory.
-	 * 
+	 *
 	 * @param internalQueueCapacity
 	 *            the size of the queue used to cache files to be received
 	 *            internally. This queue can be made larger to optimize the
@@ -124,7 +144,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	 * size of the queue is mutually exclusive with ordering. No guarantees
 	 * about file delivery order can be made under concurrent access.
 	 * <p/>
-	 * 
+	 *
 	 * @param receptionOrderComparator
 	 *            the comparator to be used to order the files in the internal
 	 *            queue
@@ -137,7 +157,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 
 	/**
 	 * Specify the input directory.
-	 * 
+	 *
 	 * @param directory to monitor
 	 */
 	public void setDirectory(File directory) {
@@ -148,7 +168,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	/**
 	 * Optionally specify a custom scanner, for example the
 	 * {@link org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner}
-	 * 
+	 *
 	 * @param scanner scanner implementation
 	 */
 	public void setScanner(DirectoryScanner scanner) {
@@ -161,7 +181,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	 * <emphasis>true</emphasis>. If set to <emphasis>false</emphasis> and the
 	 * source directory does not exist, an Exception will be thrown upon
 	 * initialization.
-	 * 
+	 *
 	 * @param autoCreateDirectory
 	 *            should the directory to be monitored be created when this
 	 *            component starts up?
@@ -180,7 +200,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	 * can be used to group them together.
 	 * <p/>
 	 * <b>The supplied filter must be thread safe.</b>.
-	 * 
+	 *
 	 * @param filter a filter
 	 */
 	public void setFilter(FileListFilter<File> filter) {
@@ -193,7 +213,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	 * duplicate processing.
 	 * <p/>
 	 * <b>The supplied FileLocker must be thread safe</b>
-	 * 
+	 *
 	 * @param locker a locker
 	 */
 	public void setLocker(FileLocker locker) {
@@ -212,7 +232,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	 * will more likely be out of sync with the file system if this flag is set
 	 * to <code>false</code>, but it will change more often (causing expensive
 	 * reordering) if it is set to <code>true</code>.
-	 * 
+	 *
 	 * @param scanEachPoll
 	 *            whether or not the component should re-scan (as opposed to not
 	 *            rescanning until the entire backlog has been delivered)
@@ -221,10 +241,27 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 		this.scanEachPoll = scanEachPoll;
 	}
 
+	public void setDispositionExpression(Expression dispositionExpression) {
+		Assert.notNull(dispositionExpression, "'dispositionExpression' must not be null");
+		this.dispositionExpression = dispositionExpression;
+	}
+
+	public void setDispositionResultChannel(MessageChannel dispositionResultChannel) {
+		Assert.notNull(dispositionResultChannel, "'dispositionResultChannel' must not be null");
+		this.dispositionMessagingTemplate.setDefaultChannel(dispositionResultChannel);
+		this.dispostionResultChannelSet = true;
+	}
+
+	public void setDispositionSendTimeout(long dispositionSendTimeout) {
+		this.dispositionMessagingTemplate.setSendTimeout(dispositionSendTimeout);
+	}
+
+	@Override
 	public String getComponentType() {
 		return "file:inbound-channel-adapter";
 	}
 
+	@Override
 	protected void onInit() {
 		Assert.notNull(directory, "'directory' must not be null");
 		if (!this.directory.exists() && this.autoCreateDirectory) {
@@ -236,6 +273,10 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 				"Source path [" + this.directory + "] does not point to a directory.");
 		Assert.isTrue(this.directory.canRead(),
 				"Source directory [" + this.directory + "] is not readable.");
+		if (getBeanFactory() != null) {
+			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(
+					new BeanFactoryResolver(getBeanFactory()));
+		}
 	}
 
 	public Message<File> receive() throws MessagingException {
@@ -259,6 +300,11 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 			if (logger.isInfoEnabled()) {
 				logger.info("Created message: [" + message + "]");
 			}
+			FileMessageHolder resource = this.resources.get();
+			if (resource == null) {
+				this.resources.set(new FileMessageHolder());
+			}
+			this.resources.get().setMessage(message);
 		}
 		return message;
 	}
@@ -276,7 +322,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 
 	/**
 	 * Adds the failed message back to the 'toBeReceived' queue if there is room.
-	 * 
+	 *
 	 * @param failedMessage
 	 *            the {@link org.springframework.integration.Message} that failed
 	 */
@@ -290,7 +336,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 	/**
 	 * The message is just logged. It was already removed from the queue during
 	 * the call to <code>receive()</code>
-	 * 
+	 *
 	 * @param sentMessage
 	 *            the message that was successfully delivered
 	 */
@@ -300,4 +346,59 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
 		}
 	}
 
+	public FileMessageHolder getResource() {
+		FileMessageHolder resource = new FileMessageHolder();
+		this.resources.set(resource);
+		return resource;
+	}
+
+	public void afterCommit(FileMessageHolder resource) {
+		Assert.isInstanceOf(FileMessageHolder.class, resource);
+		FileMessageHolder fileResource = resource;
+		if (this.dispositionExpression != null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Executing expression " + this.dispositionExpression.getExpressionString() + " on " +
+						fileResource.getMessage());
+			}
+			Object result = this.dispositionExpression.getValue(this.evaluationContext, fileResource.getMessage());
+			if (result != null) {
+				if (this.dispostionResultChannelSet) {
+					try {
+						Message<File> message = MessageBuilder.fromMessage(fileResource.getMessage())
+								.setHeader(FileHeaders.DISPOSITION_RESULT, result).build();
+						this.dispositionMessagingTemplate.send(message);
+					}
+					catch (Exception e) {
+						logger.error("Error sending File Disposition Result", e);
+					}
+				}
+			}
+		}
+		this.resources.set(null);
+	}
+
+	public void afterRollback(FileMessageHolder resource) {
+		// no op
+	}
+
+	public void afterReceiveNoTx(FileMessageHolder resource) {
+		// no op
+	}
+
+	public void afterSendNoTx(FileMessageHolder resource) {
+		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/FileInboundChannelAdapterParser.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.
@@ -16,39 +16,47 @@
 
 package org.springframework.integration.file.config;
 
-import org.w3c.dom.Element;
-
 import org.springframework.beans.BeanMetadataElement;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.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;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
+import org.w3c.dom.Element;
 
 /**
  * Parser for the <inbound-channel-adapter> element of the 'file' namespace.
- * 
+ *
  * @author Iwein Fuld
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public class FileInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
 
-	private static final String PACKAGE_NAME = "org.springframework.integration.file";
-
-
 	@Override
 	protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
-				PACKAGE_NAME + ".config.FileReadingMessageSourceFactoryBean");
+		BeanDefinitionBuilder builder =
+				BeanDefinitionBuilder.genericBeanDefinition(FileReadingMessageSourceFactoryBean.class);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "comparator");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "scanner");
 		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");
 		String filterBeanName = this.registerFilter(element, parserContext);
 		String lockerBeanName = registerLocker(element, parserContext);
 		if (lockerBeanName != null) {
@@ -64,8 +72,8 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann
 		String lockerBeanName = null;
 		Element nioLocker = DomUtils.getChildElementByTagName(element, "nio-locker");
 		if (nioLocker != null) {
-			BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
-					PACKAGE_NAME + ".locking.NioFileLocker");
+			BeanDefinitionBuilder builder =
+					BeanDefinitionBuilder.genericBeanDefinition(NioFileLocker.class);
 			lockerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
 					builder.getBeanDefinition(), parserContext.getRegistry());
 		}
@@ -79,8 +87,8 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann
 	}
 
 	private String registerFilter(Element element, ParserContext parserContext) {
-		BeanDefinitionBuilder factoryBeanBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-				PACKAGE_NAME + ".config.FileListFilterFactoryBean");
+		BeanDefinitionBuilder factoryBeanBuilder =
+				BeanDefinitionBuilder.genericBeanDefinition(FileListFilterFactoryBean.class);
 		factoryBeanBuilder.setRole(BeanDefinition.ROLE_SUPPORT);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(factoryBeanBuilder, element, "filter");
 		String filenamePattern = element.getAttribute("filename-pattern");

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

@@ -21,8 +21,9 @@ import java.util.Comparator;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.beans.factory.FactoryBean;
+import org.springframework.expression.Expression;
+import org.springframework.integration.MessageChannel;
 import org.springframework.integration.file.DirectoryScanner;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
@@ -56,6 +57,12 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean<FileRead
 
 	private volatile Integer queueSize;
 
+	private volatile Expression dispositionExpression;
+
+	private volatile MessageChannel dispositionResultChannel;
+
+	private volatile Long dispositionSendTimeout;
+
 	private final Object initializationMonitor = new Object();
 
 
@@ -94,6 +101,18 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean<FileRead
 		this.locker = locker;
 	}
 
+	public void setDispositionExpression(Expression dispositionExpression) {
+		this.dispositionExpression = dispositionExpression;
+	}
+
+	public void setDispositionResultChannel(MessageChannel dispositionResultChannel) {
+		this.dispositionResultChannel = dispositionResultChannel;
+	}
+
+	public void setDispositionSendTimeout(Long dispositionSendTimeout) {
+		this.dispositionSendTimeout = dispositionSendTimeout;
+	}
+
 	public FileReadingMessageSource getObject() throws Exception {
 		if (this.source == null) {
 			initSource();
@@ -150,6 +169,15 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean<FileRead
 			if (this.autoCreateDirectory != null) {
 				this.source.setAutoCreateDirectory(this.autoCreateDirectory);
 			}
+			if (this.dispositionExpression != null) {
+				this.source.setDispositionExpression(this.dispositionExpression);
+			}
+			if (this.dispositionResultChannel != null) {
+				this.source.setDispositionResultChannel(this.dispositionResultChannel);
+			}
+			if (this.dispositionSendTimeout != null) {
+				this.source.setDispositionSendTimeout(this.dispositionSendTimeout);
+			}
 			this.source.afterPropertiesSet();
 		}
 	}

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

@@ -166,6 +166,42 @@ Only files matching this regular expression will be picked up by this adapter.
                   </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:complexType>
     </xsd:element>
 

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

@@ -16,20 +16,32 @@
 
 package org.springframework.integration.file;
 
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.runners.MockitoJUnitRunner;
-import org.springframework.integration.Message;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.io.IOException;
 import java.util.Comparator;
 
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+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;
 
 /**
  * @author Iwein Fuld
@@ -147,4 +159,18 @@ public class FileReadingMessageSourceTests {
         assertNull(source.receive());
         verify(inputDirectoryMock, times(2)).listFiles();
     }
+
+    @Test
+    public void disposition() {
+        source.setDispositionExpression(new LiteralExpression("foo"));
+        QueueChannel channel = new QueueChannel();
+        source.setDispositionResultChannel(channel);
+        FileMessageHolder resource = source.getResource();
+        File file = mock(File.class);
+        resource.setMessage(new GenericMessage<File>(file));
+        source.afterCommit(resource);
+        Message<?> result = channel.receive(10000);
+        assertSame(file, result.getPayload());
+        assertEquals("foo", result.getHeaders().get(FileHeaders.DISPOSITION_RESULT));
+    }
 }

spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml

@@ -12,6 +12,8 @@
     <!-- under test -->
     <file:inbound-channel-adapter
             directory="#{inputDirectory.path}"
+            disposition-expression="payload.delete()"
+            disposition-result-channel="resultChannel"
             channel="fileMessages" filter="compositeFilter"/>
 
     <bean id="temp" class="org.junit.rules.TemporaryFolder"
@@ -39,4 +41,7 @@
 
     <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
 
+	<si:channel id="resultChannel">
+		<si:queue />
+	</si:channel>
 </beans>

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

@@ -16,6 +16,13 @@
 
 package org.springframework.integration.file;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -24,11 +31,6 @@ import org.springframework.integration.core.PollableChannel;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import java.io.File;
-
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-
 /**
  * @author Iwein Fuld
  */
@@ -41,15 +43,23 @@ public class FileToChannelIntegrationTests {
 	@Autowired
 	PollableChannel fileMessages;
 
+	@Autowired
+	PollableChannel resultChannel;
+
 	@Test(timeout = 2000)
 	public void fileMessageToChannel() throws Exception {
-		File.createTempFile("test", null, inputDirectory).setLastModified(System.currentTimeMillis() - 1000);
+		File file = File.createTempFile("test", null, inputDirectory);
+		file.setLastModified(System.currentTimeMillis() - 1000);
 		Message<File> received = receiveFileMessage();
 		while (received == null) {
 			Thread.sleep(50);
 			received = receiveFileMessage();
 		}
 		assertNotNull(received.getPayload());
+		Message<?> result = resultChannel.receive(10000);
+		assertNotNull(result);
+		assertEquals(Boolean.TRUE, result.getHeaders().get(FileHeaders.DISPOSITION_RESULT));
+		assertTrue(!file.exists());
 	}
 
 	@SuppressWarnings("unchecked")

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml

@@ -14,10 +14,15 @@
                              directory="${java.io.tmpdir}"
                              filter="filter"
                              comparator="testComparator"
+                             disposition-expression="payload.delete()"
+                             disposition-result-channel="resultChannel"
+                             disposition-send-timeout="123"
                              auto-startup="false">
         <integration:poller fixed-rate="5000"/>
     </inbound-channel-adapter>
 
+	<integration:channel id="resultChannel" />
+
     <beans:bean id="filter" class="org.springframework.integration.file.config.FileListFilterFactoryBean"/>
 
 	<beans:bean id="compositeFilter" class="org.springframework.integration.file.filters.CompositeFileListFilter">

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java

@@ -27,14 +27,16 @@ import java.util.concurrent.PriorityBlockingQueue;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpression;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.file.DefaultDirectoryScanner;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -92,6 +94,14 @@ public class FileInboundChannelAdapterParserTests {
         assertSame("comparator reference not set, ", expected, actual);
     }
 
+    @Test
+    public void disposition() throws Exception {
+        Object dispositionExpression = accessor.getPropertyValue("dispositionExpression");
+        assertEquals(SpelExpression.class, dispositionExpression.getClass());
+        assertEquals("payload.delete()", ((Expression) dispositionExpression).getExpressionString());
+        assertSame(TestUtils.getPropertyValue(source, "dispositionMessagingTemplate.defaultChannel"), context.getBean("resultChannel"));
+        assertEquals(123L, TestUtils.getPropertyValue(source, "dispositionMessagingTemplate.sendTimeout"));
+    }
 
     static class TestComparator implements Comparator<File> {
 

spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java

@@ -28,11 +28,8 @@ import javax.servlet.http.HttpServletResponse;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.context.expression.BeanFactoryResolver;
-import org.springframework.context.expression.MapAccessor;
-import org.springframework.core.convert.ConversionService;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
-import org.springframework.expression.spel.support.StandardTypeConverter;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
@@ -59,6 +56,7 @@ import org.springframework.integration.http.multipart.MultipartHttpInputMessage;
 import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
 import org.springframework.integration.mapping.HeaderMapper;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.util.ExpressionUtils;
 import org.springframework.util.AntPathMatcher;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -502,17 +500,13 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor
 	}
 
 	protected StandardEvaluationContext createEvaluationContext(){
-		StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
-		evaluationContext.addPropertyAccessor(new MapAccessor());
-		BeanFactory beanFactory = this.getBeanFactory();
-		if (beanFactory != null) {
-			evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory));
+		if (this.getBeanFactory() != null) {
+			return ExpressionUtils.createStandardEvaluationContext(new BeanFactoryResolver(this.getBeanFactory()),
+					this.getConversionService());
 		}
-		ConversionService conversionService = this.getConversionService();
-		if (conversionService != null) {
-			evaluationContext.setTypeConverter(new StandardTypeConverter(conversionService));
+		else {
+			return ExpressionUtils.createStandardEvaluationContext(this.getConversionService());
 		}
-		return evaluationContext;
 	}
 
 	private void validateSupportedMethods() {

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java

@@ -60,7 +60,7 @@ public class PseudoTransactionalMessageSourceTests {
 		assertTrue(rolledBack);
 	}
 
-	public static class MessageSource implements PseudoTransactionalMessageSource<String> {
+	public static class MessageSource implements PseudoTransactionalMessageSource<String, Object> {
 
 		public Message<String> receive() {
 			if (doRollback) {
@@ -83,5 +83,11 @@ public class PseudoTransactionalMessageSourceTests {
 			latch2.countDown();
 		}
 
+		public void afterReceiveNoTx(Object resource) {
+		}
+
+		public void afterSendNoTx(Object resource) {
+		}
+
 	}
 }

spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java

@@ -39,7 +39,7 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Gary Russell
  */
-public class MailReceivingMessageSource implements PseudoTransactionalMessageSource<javax.mail.Message> {
+public class MailReceivingMessageSource implements PseudoTransactionalMessageSource<javax.mail.Message, MailReceiverContext> {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
@@ -77,17 +77,31 @@ public class MailReceivingMessageSource implements PseudoTransactionalMessageSou
 		return null;
 	}
 
-	public Object getResource() {
+	public MailReceiverContext getResource() {
 		return this.mailReceiver.getTransactionContext();
 	}
 
-	public void afterCommit(Object context) {
+	public void afterCommit(MailReceiverContext context) {
 		Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
-		this.mailReceiver.closeContextAfterSuccess((MailReceiverContext) context);
+		this.mailReceiver.closeContextAfterSuccess(context);
 	}
 
-	public void afterRollback(Object context) {
+	public void afterRollback(MailReceiverContext context) {
 		Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
-		this.mailReceiver.closeContextAfterFailure((MailReceiverContext) context);
+		this.mailReceiver.closeContextAfterFailure(context);
+	}
+
+	/**
+	 * For backwards-compatibility; the mail adapter updates the status before the send.
+	 */
+	public void afterReceiveNoTx(MailReceiverContext resource) {
+		this.afterCommit(resource);
+	}
+
+	/**
+	 * For backwards-compatibility; the mail adapter updates the status before the send.
+	 */
+	public void afterSendNoTx(MailReceiverContext resource) {
+		// No op
 	}
 }