Merge pull request #564 from garyrussell/INT-2685

* INT-2685:
  INT-2685 Transaction Synchronization

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

@@ -34,7 +34,7 @@ import org.apache.commons.logging.LogFactory;
 
 /**
  * The headers for a {@link Message}.<br>
- * IMPORTANT: MessageHeaders are immutable. Any mutating operation (e.g., put(..), putAll(..) etc.) 
+ * IMPORTANT: MessageHeaders are immutable. Any mutating operation (e.g., put(..), putAll(..) etc.)
  * will result in {@link UnsupportedOperationException}
  * To create MessageHeaders instance use fluent MessageBuilder API
  * <pre>
@@ -47,17 +47,18 @@ import org.apache.commons.logging.LogFactory;
  * headers.put("key2", "value2");
  * new GenericMessage("foo", headers);
  * </pre>
- * 
+ *
  * @author Arjen Poutsma
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public final class MessageHeaders implements Map<String, Object>, Serializable {
 
 	private static final long serialVersionUID = 6901029029524535147L;
 
 	private static final Log logger = LogFactory.getLog(MessageHeaders.class);
-	
+
 	private static volatile IdGenerator idGenerator = null;
 
 	/**
@@ -88,6 +89,8 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 
 	public static final String CONTENT_TYPE = "content-type";
 
+	public static final String DISPOSITION_RESULT = "dispositionResult";
+
 
 	private final Map<String, Object> headers;
 
@@ -100,7 +103,7 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 		else {
 			this.headers.put(ID, MessageHeaders.idGenerator.generateId());
 		}
-		
+
 		this.headers.put(TIMESTAMP, new Long(System.currentTimeMillis()));
 	}
 
@@ -155,10 +158,12 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 		return (T) value;
 	}
 
+	@Override
 	public int hashCode() {
 		return this.headers.hashCode();
 	}
 
+	@Override
 	public boolean equals(Object obj) {
 		if (this == obj) {
 			return true;
@@ -170,6 +175,7 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 		return false;
 	}
 
+	@Override
 	public String toString() {
 		return this.headers.toString();
 	}

spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java

@@ -24,6 +24,7 @@ import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.SmartLifecycle;
+import org.springframework.expression.Expression;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageSource;
@@ -133,7 +134,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 			}
 			if (this.pollerMetadata.getMaxMessagesPerPoll() == Integer.MIN_VALUE){
 				// the default is 1 since a source might return
-				// a non-null and non-interruptable value every time it is invoked
+				// a non-null and non-interruptible value every time it is invoked
 				this.pollerMetadata.setMaxMessagesPerPoll(1);
 			}
 			spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
@@ -144,7 +145,26 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 			spca.setAdviceChain(this.pollerMetadata.getAdviceChain());
 			spca.setTrigger(this.pollerMetadata.getTrigger());
 			spca.setErrorHandler(this.pollerMetadata.getErrorHandler());
-			spca.setSynchronized(this.pollerMetadata.isSynchronized());
+			Expression onSuccessExpression = this.pollerMetadata.getOnSuccessExpression();
+			if (onSuccessExpression != null) {
+				spca.setOnSuccessExpression(onSuccessExpression);
+			}
+			MessageChannel onSuccessResultChannel = this.pollerMetadata.getOnSuccessResultChannel();
+			if (onSuccessResultChannel != null) {
+				spca.setOnSuccessResultChannel(onSuccessResultChannel);
+			}
+			Expression onFailureExpression = this.pollerMetadata.getOnFailureExpression();
+			if (onFailureExpression != null) {
+				spca.setOnFailureExpression(onFailureExpression);
+			}
+			MessageChannel onFailureResultChannel = this.pollerMetadata.getOnFailureResultChannel();
+			if (onFailureResultChannel != null) {
+				spca.setOnFailureChannel(onFailureResultChannel);
+			}
+			Long resultSendTimeout = this.pollerMetadata.getSendTimeout();
+			if (resultSendTimeout != null) {
+				spca.setResultSendTimeout(resultSendTimeout);
+			}
 			spca.setBeanClassLoader(this.beanClassLoader);
 			spca.setAutoStartup(this.autoStartup);
 			spca.setBeanName(this.beanName);

spring-integration-core/src/main/java/org/springframework/integration/config/xml/PollerParser.java

@@ -28,9 +28,11 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.support.ManagedList;
+import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
+import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.scheduling.support.CronTrigger;
@@ -48,6 +50,7 @@ import org.w3c.dom.NodeList;
  * @author Marius Bogoevici
  * @author Oleg Zhurakousky
  * @author Artem Bilan
+ * @author Gary Russell
  */
 public class PollerParser extends AbstractBeanDefinitionParser {
 
@@ -93,6 +96,19 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 		Element adviceChainElement = DomUtils.getChildElementByTagName(element, "advice-chain");
 		configureAdviceChain(adviceChainElement, txElement, metadataBuilder, parserContext);
 
+		Element pseudoTxElement = DomUtils.getChildElementByTagName(element, "psuedo-transactional");
+		if (pseudoTxElement != null && txElement != null) {
+			parserContext.getReaderContext().error(
+					"Cannot have both 'transactional' and 'pseudo-transactional' elements", element);
+		}
+		Element txSyncElement = DomUtils.getChildElementByTagName(element, "transaction-synchronization");
+		if (pseudoTxElement != null && txSyncElement != null) {
+			parserContext.getReaderContext().error(
+					"Cannot have both 'transaction-synchronization' and 'pseudo-transactional' elements", element);
+		}
+		pseudoTxElement = pseudoTxElement == null ? txSyncElement : pseudoTxElement;
+		configureTransactionSync(pseudoTxElement, metadataBuilder, parserContext);
+
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "task-executor");
 		String errorChannel = element.getAttribute("error-channel");
 		if (StringUtils.hasText(errorChannel)) {
@@ -100,7 +116,6 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 			errorHandler.addPropertyReference("defaultErrorChannel", errorChannel);
 			metadataBuilder.addPropertyValue("errorHandler", errorHandler.getBeanDefinition());
 		}
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(metadataBuilder, element, "synchronized");
 		return metadataBuilder.getBeanDefinition();
 	}
 
@@ -201,4 +216,24 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 		targetBuilder.addPropertyValue("adviceChain", adviceChain);
 	}
 
+	private void configureTransactionSync(Element element, BeanDefinitionBuilder metadataBuilder,
+			ParserContext parserContext) {
+		if (element != null) {
+			configureSyncExpression(element, metadataBuilder, parserContext, "on-success-expression", "onSuccessExpression");
+			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "on-success-result-channel");
+			configureSyncExpression(element, metadataBuilder, parserContext, "on-failure-expression", "onFailureExpression");
+			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "on-failure-result-channel");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(metadataBuilder, element, "send-timeout", "sendTimeout");
+		}
+	}
+
+	private void configureSyncExpression(Element element, BeanDefinitionBuilder metadataBuilder,
+			ParserContext parserContext, String expressionAttribute, String expressionProperty) {
+		String expression = element.getAttribute(expressionAttribute);
+		if (StringUtils.hasText(expression)) {
+			RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
+			expressionDef.getConstructorArgumentValues().addGenericArgumentValue(expression);
+			metadataBuilder.addPropertyValue(expressionProperty, expressionDef);
+		}
+	}
 }

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

@@ -34,6 +34,10 @@ import org.springframework.transaction.support.TransactionSynchronization;
  * small (but present) window in which a transaction might commit but the
  * resource is not updated to reflect that. This could result in
  * duplicate messages.
+ * <p>All {@link MessageSource}s can have success/failure expressions evaluated either as part
+ * of a transaction with a &lt;transactional/&gt; poller or after success/failure when
+ * running in a &lt;pseudo-transactional/&gt; poller. This interface is for those
+ * message sources that need additional flexibility than that provided by SpEL expressions.
  * @author Gary Russell
  * @since 2.2
  *
@@ -42,7 +46,9 @@ public interface PseudoTransactionalMessageSource<T, V> extends MessageSource<T>
 
 	/**
 	 * Obtain the resource on which appropriate action needs
-	 * to be taken.
+	 * to be taken. This resource is passed back into the other
+	 * methods. In addition, it is made available to transaction
+	 * synchronization SpEL expressions in the '#resource' variable.
 	 * @return The resource.
 	 */
 	V getResource();
@@ -50,16 +56,16 @@ public interface PseudoTransactionalMessageSource<T, V> extends MessageSource<T>
 	/**
 	 * Invoked via {@link TransactionSynchronization} when the
 	 * transaction commits.
-	 * @param resource The resource to be "committed"
+	 * @param object The resource to be "committed"
 	 */
-	void afterCommit(V resource);
+	void afterCommit(Object object);
 
 	/**
 	 * Invoked via {@link TransactionSynchronization} when the
 	 * transaction rolls back.
-	 * @param resource
+	 * @param object
 	 */
-	void afterRollback(V resource);
+	void afterRollback(Object object);
 
 	/**
 	 * Called when there is no transaction and the receive() call completed.

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

@@ -16,14 +16,21 @@
 
 package org.springframework.integration.endpoint;
 
+import org.springframework.context.expression.BeanFactoryResolver;
+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.MessageHeaders;
+import org.springframework.integration.MessagingException;
 import org.springframework.integration.context.NamedComponent;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.core.PseudoTransactionalMessageSource;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.history.TrackableComponent;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.util.ExpressionUtils;
 import org.springframework.transaction.support.ResourceHolder;
 import org.springframework.transaction.support.ResourceHolderSynchronization;
 import org.springframework.transaction.support.TransactionSynchronization;
@@ -40,6 +47,13 @@ import org.springframework.util.Assert;
  */
 public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent {
 
+	/**
+	 * Transaction synchronization needs a non-null resource; this constant is used for
+	 * message sources that have no need for a resource, because the post-process
+	 * action just needs the message.
+	 */
+	private static final Object NO_TX_RESOURCE = new Object();
+
 	private volatile MessageSource<?> source;
 
 	private volatile boolean isPseudoTxMessageSource;
@@ -50,7 +64,15 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
-	private volatile boolean synchronizedTx = true;
+	private volatile Expression onSuccessExpression;
+
+	private final MessagingTemplate onSuccessMessagingTemplate = new MessagingTemplate();
+
+	private volatile Expression onFailureExpression;
+
+	private final MessagingTemplate onFailureMessagingTemplate = new MessagingTemplate();
+
+	private volatile StandardEvaluationContext evaluationContext;
 
 	/**
 	 * Specify the source to be polled for Messages.
@@ -82,8 +104,30 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 		this.shouldTrack = shouldTrack;
 	}
 
-	public void setSynchronized(boolean synchronizedTx) {
-		this.synchronizedTx = synchronizedTx;
+
+	public void setOnSuccessExpression(Expression onSuccessExpression) {
+		Assert.notNull(onSuccessExpression, "onSuccessExpression cannot be null");
+		this.onSuccessExpression = onSuccessExpression;
+	}
+
+	public void setOnFailureExpression(Expression onFailureExpression) {
+		Assert.notNull(onFailureExpression, "onFailureExpression cannot be null");
+		this.onFailureExpression = onFailureExpression;
+	}
+
+	public void setOnSuccessResultChannel(MessageChannel onSuccessResultChannel) {
+		Assert.notNull(onSuccessResultChannel, "onSuccessChannel cannot be null");
+		this.onSuccessMessagingTemplate.setDefaultChannel(onSuccessResultChannel);
+	}
+
+	public void setOnFailureChannel(MessageChannel onFailureResultChannel) {
+		Assert.notNull(onFailureResultChannel, "onFailureChannel cannot be null");
+		this.onFailureMessagingTemplate.setDefaultChannel(onFailureResultChannel);
+	}
+
+	public void setResultSendTimeout(long sendTimeout) {
+		this.onSuccessMessagingTemplate.setSendTimeout(sendTimeout);
+		this.onFailureMessagingTemplate.setSendTimeout(sendTimeout);
 	}
 
 	@Override
@@ -96,6 +140,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 	protected void onInit() {
 		Assert.notNull(this.source, "source must not be null");
 		Assert.notNull(this.outputChannel, "outputChannel must not be null");
+		this.evaluationContext = this.createEvaluationContext();
 		super.onInit();
 	}
 
@@ -104,25 +149,32 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 	protected boolean doPoll() {
 		boolean isInTx = false;
 		PseudoTransactionalMessageSource<?,Object> messageSource = null;
-		Object resource = null;
+		Object resource = NO_TX_RESOURCE;
 		if (this.isPseudoTxMessageSource) {
 			messageSource = (PseudoTransactionalMessageSource<?,Object>) this.source;
 			resource = messageSource.getResource();
-			Assert.state(resource != null, "Pseudo Transactional Message Source returned null resource");
-			if (this.synchronizedTx && TransactionSynchronizationManager.isActualTransactionActive()) {
-				TransactionSynchronizationManager.bindResource(messageSource, resource);
-				TransactionSynchronizationManager.registerSynchronization(
-					new PseudoTransactionalResourceSynchronization(
-						new PseudoTransactionalResourceHolder(resource), this.source));
-				isInTx = true;
-			}
 		}
 		Message<?> message;
 		try {
 			message = this.source.receive();
+			if (TransactionSynchronizationManager.isActualTransactionActive()) {
+				TransactionSynchronizationManager.bindResource(this, resource);
+				TransactionSynchronizationManager.registerSynchronization(
+					new PseudoTransactionalResourceSynchronization(
+						new PseudoTransactionalResourceHolder(message, resource), this));
+				isInTx = true;
+			}
 		}
 		finally {
-			if (this.isPseudoTxMessageSource && !isInTx) {
+			if (!isInTx && this.isPseudoTxMessageSource) {
+				/*
+				 * This callback is provided for 'legacy' message sources
+				 * that used to take action after the receive and before
+				 * the send. When in a transaction, that action is now
+				 * taken after the commit but, when not in a transaction
+				 * this callback provides backwards compatibility. An
+				 * example is the mail reader that deletes from the inbox.
+				 */
 				messageSource.afterReceiveNoTx(resource);
 			}
 		}
@@ -133,9 +185,30 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 			if (this.shouldTrack) {
 				message = MessageHistory.write(message, this);
 			}
-			this.messagingTemplate.send(this.outputChannel, message);
-			if (this.isPseudoTxMessageSource && !isInTx) {
-				messageSource.afterSendNoTx(resource);
+			try {
+				this.messagingTemplate.send(this.outputChannel, message);
+				if (!isInTx) {
+					if (this.isPseudoTxMessageSource) {
+						/*
+						 * For 'legacy' message sources that need more flexibility than simple
+						 * expression evaluation, we invoke this callback after a
+						 * successful send.
+						 */
+						messageSource.afterSendNoTx(resource);
+					}
+					this.onSuccess(message, resource);
+				}
+			}
+			catch (Exception e) {
+				if (!isInTx) {
+					this.onFailure(message, resource);
+				}
+				if (e instanceof MessagingException) {
+					throw (MessagingException) e;
+				}
+				else {
+					throw new MessagingException(message, e);
+				}
 			}
 			return true;
 		}
@@ -145,11 +218,78 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 		return false;
 	}
 
+	private void onSuccess(Message<?> message, Object resource) {
+		doPostProcess(message, resource, this.onSuccessExpression, this.onSuccessMessagingTemplate, "success");
+	}
+
+	private void onFailure(Message<?> message, Object resource) {
+		doPostProcess(message, resource, this.onFailureExpression, this.onFailureMessagingTemplate, "failure");
+	}
+
+	private void doPostProcess(Message<?> message, Object resource, Expression expression,
+			MessagingTemplate messagingTemplate, String expressionType) {
+		if (expression != null && message != null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Evaluating " + expressionType + " expression: '" + expression.getExpressionString() + "' on " + message);
+			}
+			StandardEvaluationContext evaluationContextToUse = this.determineEvaluationContextToUse(resource);
+			Object value;
+			try {
+				value = expression.getValue(evaluationContextToUse, message);
+			}
+			catch (Exception e) {
+				value = e;
+			}
+			if (value != null) {
+				try {
+					messagingTemplate.send(MessageBuilder.fromMessage(message)
+							.setHeader(MessageHeaders.DISPOSITION_RESULT, value).build());
+				}
+				catch (Exception e) {
+					logger.error("Failed to send " + expressionType + " evaluation result " + message, e);
+				}
+			}
+		}
+	}
+
+	/**
+	 * If we don't need a resource variable (not a {@link PseudoTransactionalMessageSource})
+	 * we can use a singleton context; otherwise we need a new one each time.
+	 * @param resource The resource
+	 * @return The context.
+	 */
+	private StandardEvaluationContext determineEvaluationContextToUse(Object resource) {
+		StandardEvaluationContext evaluationContextToUse;
+		if (resource != NO_TX_RESOURCE) {
+			evaluationContextToUse = this.createEvaluationContext();
+			evaluationContextToUse.setVariable("resource", resource);
+		}
+		else {
+			if (this.evaluationContext == null) {
+				this.evaluationContext = this.createEvaluationContext();
+			}
+			evaluationContextToUse = this.evaluationContext;
+		}
+		return evaluationContextToUse;
+	}
+
+	protected StandardEvaluationContext createEvaluationContext(){
+		if (this.getBeanFactory() != null) {
+			return ExpressionUtils.createStandardEvaluationContext(new BeanFactoryResolver(this.getBeanFactory()),
+					this.getConversionService());
+		}
+		else {
+			return ExpressionUtils.createStandardEvaluationContext(this.getConversionService());
+		}
+	}
+
 	private class PseudoTransactionalResourceHolder implements ResourceHolder {
 
+		private final Message<?> message;
 		private final Object resource;
 
-		public PseudoTransactionalResourceHolder(Object resource) {
+		public PseudoTransactionalResourceHolder(Message<?> message, Object resource) {
+			this.message = message;
 			this.resource = resource;
 		}
 
@@ -157,6 +297,10 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 			return resource;
 		}
 
+		public Message<?> getMessage() {
+			return message;
+		}
+
 		public void reset() {
 		}
 
@@ -185,28 +329,30 @@ 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<?,Object>) source).afterCommit(resourceHolder.getResource());
+			if (isPseudoTxMessageSource) {
+				((PseudoTransactionalMessageSource<?, ?>) source).afterCommit(resourceHolder.getResource());
+			}
+			onSuccess(resourceHolder.getMessage(), 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<?,Object>) source).afterRollback(this.resourceHolder.getResource());
+				if (isPseudoTxMessageSource) {
+					((PseudoTransactionalMessageSource<?, ?>) source).afterRollback(resourceHolder.getResource());
+				}
+				onFailure(this.resourceHolder.getMessage(), this.resourceHolder.getResource());
 			}
 			super.afterCompletion(status);
 		}
 
-
-
 	}
 }

spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.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.
@@ -20,12 +20,15 @@ import java.util.List;
 import java.util.concurrent.Executor;
 
 import org.aopalliance.aop.Advice;
+import org.springframework.expression.Expression;
+import org.springframework.integration.MessageChannel;
 import org.springframework.scheduling.Trigger;
 import org.springframework.util.ErrorHandler;
 
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public class PollerMetadata {
 
@@ -39,11 +42,19 @@ public class PollerMetadata {
 
 	private volatile ErrorHandler errorHandler;
 
-	private List<Advice> adviceChain;
+	private volatile List<Advice> adviceChain;
 
 	private volatile Executor taskExecutor;
 
-	private volatile boolean synchronizedTx = true;
+	private volatile Expression onSuccessExpression;
+
+	private volatile MessageChannel onSuccessResultChannel;
+
+	private volatile Expression onFailureExpression;
+
+	private volatile MessageChannel onFailureResultChannel;
+
+	private volatile long sendTimeout;
 
 	public void setTrigger(Trigger trigger) {
 		this.trigger = trigger;
@@ -102,11 +113,44 @@ public class PollerMetadata {
 		return this.taskExecutor;
 	}
 
-	public boolean isSynchronized() {
-		return synchronizedTx;
+	public Expression getOnSuccessExpression() {
+		return onSuccessExpression;
 	}
 
-	public void setSynchronized(boolean synchronizedTx) {
-		this.synchronizedTx = synchronizedTx;
+	public void setOnSuccessExpression(Expression onSuccessExpression) {
+		this.onSuccessExpression = onSuccessExpression;
 	}
+
+	public MessageChannel getOnSuccessResultChannel() {
+		return onSuccessResultChannel;
+	}
+
+	public void setOnSuccessResultChannel(MessageChannel onSuccessResultChannel) {
+		this.onSuccessResultChannel = onSuccessResultChannel;
+	}
+
+	public Expression getOnFailureExpression() {
+		return onFailureExpression;
+	}
+
+	public void setOnFailureExpression(Expression onFailureExpression) {
+		this.onFailureExpression = onFailureExpression;
+	}
+
+	public MessageChannel getOnFailureResultChannel() {
+		return onFailureResultChannel;
+	}
+
+	public void setOnFailureResultChannel(MessageChannel onFailureResultChannel) {
+		this.onFailureResultChannel = onFailureResultChannel;
+	}
+
+	public long getSendTimeout() {
+		return sendTimeout;
+	}
+
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
 }

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

@@ -1504,6 +1504,34 @@
 					</xsd:complexType>
 				</xsd:element>
 			</xsd:choice>
+			<xsd:choice minOccurs="0" maxOccurs="1">
+				<xsd:element name="psuedo-transactional" type="pseudoTransactionalType" minOccurs="0" maxOccurs="1">
+					<xsd:annotation>
+						<xsd:documentation><![CDATA[
+							This element is used to provide expressions that will be executed on success or failure,
+							when sending a message resulting from a poll. It provides semantics similar to a transaction
+							but users need to be aware that it is not a true transaction because the underlying resource
+							is not transactional.
+							The element is not allowed when a <transactional/> element is present.
+						]]></xsd:documentation>
+					</xsd:annotation>
+				</xsd:element>
+				<xsd:element name="transaction-synchronization" type="pseudoTransactionalType" minOccurs="0" maxOccurs="1">
+					<xsd:annotation>
+						<xsd:documentation><![CDATA[
+							This element is used with <transactional/> to provide expressions that will be executed
+							on success or failure, when sending a message resulting from a poll.
+							The execution of the expressions is synchronized
+							with the encompassing transaction in that the onSuccess expression will be evaluated
+							immediately after the commit or the onFailure expression will be evaluated immediately
+							after the rollback. Users need to be aware that transaction synchronization does not
+							make an inherently non-transactional resource transactional, it simply implements the
+							best-effort one phase commit pattern.
+							The element is only allowed when a <transactional/> or <advice-chain/> element is present.
+						]]></xsd:documentation>
+					</xsd:annotation>
+				</xsd:element>
+			</xsd:choice>
 		</xsd:sequence>
 		<xsd:attribute name="fixed-delay" type="xsd:string">
 			<xsd:annotation>
@@ -1578,20 +1606,6 @@
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="synchronized" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation>
-					Specifies whether the resource, used by the MessageSource that this poller
-					polls, is synchronized with the transaction. The resource may be disposed of
-					in different manners, depending on whether the transaction commits,
-					or rolls back. Only applied if a transaction subelement (or
-					an advice-chain that contains a transaction advice)	is provided.
-					Also, only applies if the MessageSource implements
-					PseudoTransactionalMessageSource.
-					Default true.
-				</xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
 	</xsd:complexType>
 
 	<xsd:element name="selector-chain">
@@ -3404,6 +3418,10 @@ is provided, the return value is expected to match a channel name exactly.
 		</xsd:attribute>
 	</xsd:complexType>
 
+	<xsd:complexType name="pseudoTransactionalType">
+		<xsd:attributeGroup ref="transactionSyncAttributeGroup" />
+	</xsd:complexType>
+
 	<xsd:complexType name="expressionOrInnerEndpointDefinitionAware">
 		<xsd:complexContent>
 			<xsd:extension base="handlerEndpointType">
@@ -3742,4 +3760,72 @@ endpoint itself is a Polling Consumer for a channel with a queue.
 			</xsd:annotation>
 		</xsd:attribute>
 	</xsd:attributeGroup>
+	
+	<xsd:attributeGroup name="transactionSyncAttributeGroup">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+				Attributes provided in either a <transactional/> or <psedo-transactional/> poller
+				sub element.
+				Used to take action after the transaction completes (<transactional/>) or after
+				the channel.send() is complete (<pseudo-transactional/>). 
+			]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="on-success-expression">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Expression to be evaluated when no exception is thrown (<pseudo-transactional/>),
+					or after the transaction commits (<transactional/>). The #root variable of the
+					expression evaluation is the original message; a BeanResolver is also available. 
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="on-success-result-channel">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Channel where the result (if any) from the evaluation of the on-success-expression is sent.
+					The message sent is the original message, enhanced with a 'dispositionResult' header
+					containing the result, or an Exception, if the evaluation failed. Default:
+					nullChannel.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+				    <tool:annotation kind="ref">
+				        <tool:expected-type type="org.springframework.integration.MessageChannel"/>
+				    </tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="on-failure-expression">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Expression to be evaluated when an exception is thrown (<pseudo-transactional/>),
+					or after the transaction rolls back (<transactional/>). The #root variable of the
+					expression evaluation is the original message; a BeanResolver is also available. 
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="on-failure-result-channel">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Channel where the result (if any) from the evaluation of the on-failure-expression is sent.
+					The message sent is the original message, enhanced with a 'dispositionResult' header
+					containing the result, or an Exception, if the evaluation failed. Default:
+					nullChannel.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+				    <tool:annotation kind="ref">
+				        <tool:expected-type type="org.springframework.integration.MessageChannel"/>
+				    </tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="send-timeout">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					A timout used when sending expression evaluation results to the success or
+					failure channels. Only applies if the channel can block on a send, such as
+					a limited-capacity QueueChannel that is currently full.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
 </xsd:schema>

spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java

@@ -140,24 +140,4 @@ public class PollerParserTests {
 				"pollerWithCronAndFixedDelay.xml", PollerParserTests.class);
 	}
 
-	@Test
-	public void pollerWithSync() {
-		ApplicationContext context = new ClassPathXmlApplicationContext(
-				"pollerWithSynchronization.xml", PollerParserTests.class);
-		Object poller = context.getBean("noSync");
-		assertNotNull(poller);
-		PollerMetadata metadata = (PollerMetadata) poller;
-		assertEquals(true, metadata.isSynchronized());
-
-		poller = context.getBean("syncTrue");
-		assertNotNull(poller);
-		metadata = (PollerMetadata) poller;
-		assertEquals(true, metadata.isSynchronized());
-
-		poller = context.getBean("syncFalse");
-		assertNotNull(poller);
-		metadata = (PollerMetadata) poller;
-		assertEquals(false, metadata.isSynchronized());
-	}
-
 }

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

@@ -15,13 +15,17 @@
  */
 package org.springframework.integration.endpoint;
 
+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 java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.PseudoTransactionalMessageSource;
 import org.springframework.integration.message.GenericMessage;
@@ -76,9 +80,88 @@ public class PseudoTransactionalMessageSourceTests {
 		assertSame(object, committed.get());
 		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
 		TransactionSynchronizationManager.clearSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(false);
 		assertNull(rolledBack.get());
 	}
 
+	@Test
+	public void testPseudoCommitWithMessage() {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		final Object object = new Object();
+		final AtomicReference<Object> afterReceive = new AtomicReference<Object>();
+		final AtomicReference<Object> afterSend = new AtomicReference<Object>();
+		adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
+
+			public Message<String> receive() {
+				return new GenericMessage<String>("foo");
+			}
+
+			public Object getResource() {
+				return object;
+			}
+
+			public void afterCommit(Object resource) {
+				throw new RuntimeException("no tx - commit not expected");
+			}
+
+			public void afterRollback(Object resource) {
+				throw new RuntimeException("no tx - rollback not expected");
+			}
+
+			public void afterReceiveNoTx(Object resource) {
+				afterReceive.set(resource);
+			}
+
+			public void afterSendNoTx(Object resource) {
+				afterSend.set(resource);
+			}
+		});
+
+		adapter.doPoll();
+		assertSame(object, afterReceive.get());
+		assertSame(object, afterSend.get());
+	}
+
+	@Test
+	public void testPseudoCommitNoMessage() {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		final Object object = new Object();
+		final AtomicReference<Object> afterReceive = new AtomicReference<Object>();
+		adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
+
+			public Message<String> receive() {
+				return null;
+			}
+
+			public Object getResource() {
+				return object;
+			}
+
+			public void afterCommit(Object resource) {
+				throw new RuntimeException("no tx - commit not expected");
+			}
+
+			public void afterRollback(Object resource) {
+				throw new RuntimeException("no tx - rollback not expected");
+			}
+
+			public void afterReceiveNoTx(Object resource) {
+				afterReceive.set(resource);
+			}
+
+			public void afterSendNoTx(Object resource) {
+				throw new RuntimeException("no message - after send not expected");
+			}
+		});
+
+		adapter.doPoll();
+		assertSame(object, afterReceive.get());
+	}
+
 	@Test
 	public void testRollback() {
 		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
@@ -118,7 +201,80 @@ public class PseudoTransactionalMessageSourceTests {
 		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
 		assertSame(object, rolledBack.get());
 		TransactionSynchronizationManager.clearSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(false);
 		assertNull(committed.get());
 	}
 
+	@Test
+	public void testSuccessAndFailureEvaluationWithResource() {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		final Object object = new Bar();
+		final AtomicReference<Object> committed = new AtomicReference<Object>();
+		final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
+		adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
+
+			public Message<String> receive() {
+				return new GenericMessage<String>("foo");
+			}
+
+			public Object getResource() {
+				return object;
+			}
+
+			public void afterCommit(Object resource) {
+				committed.set(resource);
+			}
+
+			public void afterRollback(Object resource) {
+				rolledBack.set(resource);
+			}
+
+			public void afterReceiveNoTx(Object resource) {
+			}
+
+			public void afterSendNoTx(Object resource) {
+			}
+		});
+
+		TransactionSynchronizationManager.initSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(true);
+		adapter.setOnSuccessExpression(new SpelExpressionParser().parseExpression("payload + #resource.value"));
+		QueueChannel success = new QueueChannel();
+		adapter.setOnSuccessResultChannel(success);
+		adapter.setOnFailureExpression(new SpelExpressionParser().parseExpression("payload + 'X' + #resource.value"));
+		QueueChannel failure = new QueueChannel();
+		adapter.setOnFailureChannel(failure);
+
+		adapter.doPoll();
+		TransactionSynchronizationUtils.triggerAfterCommit();
+		assertSame(object, committed.get());
+		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+		TransactionSynchronizationManager.clearSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(false);
+		assertNull(rolledBack.get());
+		Message<?> result = success.receive(10000);
+		assertNotNull(result);
+		assertEquals("foobar", result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
+		committed.set(null);
+
+		TransactionSynchronizationManager.initSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(true);
+		adapter.doPoll();
+		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+		assertSame(object, rolledBack.get());
+		TransactionSynchronizationManager.clearSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(false);
+		assertNull(committed.get());
+		result = failure.receive(10000);
+		assertNotNull(result);
+		assertEquals("fooXbar", result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
+	}
+
+	public class Bar {
+		public String getValue() {
+			return "bar";
+		}
+	}
 }

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

@@ -35,6 +35,4 @@ 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

@@ -26,22 +26,14 @@ 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.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;
 
 /**
@@ -73,7 +65,7 @@ import org.springframework.util.Assert;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  */
-public class FileReadingMessageSource extends IntegrationObjectSupport implements PseudoTransactionalMessageSource<File, FileMessageHolder> {
+public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource<File> {
 
 	private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5;
 
@@ -95,17 +87,8 @@ 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.
 	 */
@@ -240,21 +223,6 @@ 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";
@@ -272,10 +240,6 @@ 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 {
@@ -345,47 +309,4 @@ 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);
-	}
-
 }

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

@@ -70,7 +70,6 @@ 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

@@ -47,7 +47,6 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size");
-		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

@@ -1,43 +0,0 @@
-/*
- * 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/config/FileReadingMessageSourceFactoryBean.java

@@ -22,8 +22,6 @@ 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;
@@ -57,12 +55,6 @@ 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();
 
 
@@ -101,18 +93,6 @@ 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();
@@ -169,15 +149,6 @@ 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/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java

@@ -22,13 +22,10 @@ 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.PseudoTransactionalMessageSource;
+import org.springframework.integration.core.MessageSource;
 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;
@@ -59,7 +56,7 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  */
 public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends MessageProducerSupport
-	implements PseudoTransactionalMessageSource<File, FileMessageHolder> {
+	implements MessageSource<File> {
 
 	/**
 	 * Should the endpoint attempt to create the local directory? True by default.
@@ -107,18 +104,6 @@ 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");
@@ -172,24 +157,4 @@ 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/main/resources/org/springframework/integration/file/config/spring-integration-file-2.2.xsd

@@ -166,42 +166,6 @@ 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/FileInboundTransactionTests-context.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int-file="http://www.springframework.org/schema/integration/file"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file-2.2.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<context:property-placeholder/>
+
+    <int-file:inbound-channel-adapter id="pseudoTx" channel="input" auto-startup="false"
+                             directory="${java.io.tmpdir}/si-test1">
+        <int:poller fixed-rate="500">
+        	<int:psuedo-transactional on-success-expression="payload.delete()"
+        				on-success-result-channel="successChannel"
+        				on-failure-expression="'foo'"
+        				on-failure-result-channel="failureChannel"
+        				send-timeout="500" />
+        </int:poller>
+    </int-file:inbound-channel-adapter>
+
+	<int:channel id="input" />
+	
+	<int:channel id="successChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int:channel id="failureChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int:channel id="txInput" />
+	
+    <int-file:inbound-channel-adapter id="realTx" channel="txInput" auto-startup="false"
+                             directory="${java.io.tmpdir}/si-test2">
+        <int:poller fixed-rate="500">
+        	<int:transactional transaction-manager="txManager" />
+        	<int:transaction-synchronization on-success-expression="@txManager.committed"
+        				on-success-result-channel="successChannel"
+        				on-failure-expression="@txManager.rolledBack"
+        				on-failure-result-channel="failureChannel"
+        				send-timeout="5000" />
+        </int:poller>
+    </int-file:inbound-channel-adapter>
+
+	<bean id="txManager" class="org.springframework.integration.file.FileInboundTransactionTests$DummyTxManager" />
+
+</beans>

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

@@ -0,0 +1,194 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageHeaders;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.core.SubscribableChannel;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionException;
+import org.springframework.transaction.support.AbstractPlatformTransactionManager;
+import org.springframework.transaction.support.DefaultTransactionStatus;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class FileInboundTransactionTests {
+
+	@Autowired
+	private SourcePollingChannelAdapter pseudoTx;
+
+	@Autowired
+	private SourcePollingChannelAdapter realTx;
+
+	@Autowired
+	private SubscribableChannel input;
+
+	@Autowired
+	private SubscribableChannel txInput;
+
+	@Autowired
+	private PollableChannel successChannel;
+
+	@Autowired
+	private PollableChannel failureChannel;
+
+	@Autowired
+	private DummyTxManager transactionManager;
+
+	@Value("${java.io.tmpdir}")
+	private String tmpDir;
+
+	@Test
+	public void testNoTx() throws Exception {
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicBoolean crash = new AtomicBoolean();
+		input.subscribe(new MessageHandler() {
+
+			public void handleMessage(Message<?> message) throws MessagingException {
+				System.out.println(message);
+				if (crash.get()) {
+					throw new MessagingException("eek");
+				}
+				latch.countDown();
+			}
+		});
+		pseudoTx.start();
+		new File(tmpDir + "/si-test1").mkdir();
+		File file = new File(tmpDir + "/si-test1/foo");
+		file.createNewFile();
+		Message<?> result = successChannel.receive(10000);
+		assertNotNull(result);
+		assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
+		System.out.println(result);
+		assertFalse(file.delete());
+		crash.set(true);
+		file = new File(tmpDir + "/si-test1/bar");
+		file.createNewFile();
+		result = failureChannel.receive(10000);
+		assertNotNull(result);
+		System.out.println(result);
+		assertTrue(file.delete());
+		assertEquals("foo", result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
+		pseudoTx.stop();
+		assertFalse(transactionManager.getCommitted());
+		assertFalse(transactionManager.getRolledBack());
+	}
+
+	@Test
+	public void testTx() throws Exception {
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicBoolean crash = new AtomicBoolean();
+		txInput.subscribe(new MessageHandler() {
+
+			public void handleMessage(Message<?> message) throws MessagingException {
+				System.out.println(message);
+				if (crash.get()) {
+					throw new MessagingException("eek");
+				}
+				latch.countDown();
+			}
+		});
+		realTx.start();
+		new File(tmpDir + "/si-test2").mkdir();
+		File file = new File(tmpDir + "/si-test2/baz");
+		file.createNewFile();
+		Message<?> result = successChannel.receive(10000);
+		assertNotNull(result);
+		assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
+		assertTrue(file.delete());
+		System.out.println(result);
+		assertTrue(transactionManager.getCommitted());
+		crash.set(true);
+		file = new File(tmpDir + "/si-test2/qux");
+		file.createNewFile();
+		result = failureChannel.receive(10000);
+		assertNotNull(result);
+		System.out.println(result);
+		assertTrue(file.delete());
+		assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
+		realTx.stop();
+		assertTrue(transactionManager.getRolledBack());
+	}
+
+	public static class DummyTxManager extends AbstractPlatformTransactionManager {
+
+		private static final long serialVersionUID = 1L;
+
+		boolean committed;
+
+		boolean rolledBack;
+
+		@Override
+		protected Object doGetTransaction() throws TransactionException {
+			return new Object();
+		}
+
+		@Override
+		protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
+		}
+
+		@Override
+		protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+			committed = true;
+		}
+
+		@Override
+		protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
+			rolledBack = true;
+		}
+
+		/**
+		 * Evaluated in transactional onSuccessExpression - ensures we rolled back before evaluation
+		 * @return
+		 */
+		public boolean getCommitted() {
+			return committed;
+		}
+
+		/**
+		 * Evaluated in transactional onFailureExpression - ensures we rolled back before evaluation
+		 * @return
+		 */
+		public boolean getRolledBack() {
+			return rolledBack;
+		}
+	}
+
+}

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

@@ -37,10 +37,7 @@ 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.message.GenericMessage;
 
 /**
  * @author Iwein Fuld
@@ -159,17 +156,4 @@ public class FileReadingMessageSourceTests {
         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,8 +12,6 @@
     <!-- 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"
@@ -37,7 +35,10 @@
 		</constructor-arg>
 	</bean>
 
-    <si:poller default="true" fixed-rate="10"/>
+	<si:poller default="true" fixed-rate="10">
+		<si:psuedo-transactional on-success-expression="payload.delete()"
+			on-success-result-channel="resultChannel" />
+	</si:poller>
 
     <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
 

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

@@ -27,6 +27,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -58,7 +59,7 @@ public class FileToChannelIntegrationTests {
 		assertNotNull(received.getPayload());
 		Message<?> result = resultChannel.receive(10000);
 		assertNotNull(result);
-		assertEquals(Boolean.TRUE, result.getHeaders().get(FileHeaders.DISPOSITION_RESULT));
+		assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
 		assertTrue(!file.exists());
 	}
 

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

@@ -14,14 +14,17 @@
                              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"/>
+        <integration:poller fixed-rate="5000">
+        	<integration:psuedo-transactional on-success-expression="payload.delete()"
+        				on-success-result-channel="successChannel"
+        				on-failure-expression="'foo'"
+        				on-failure-result-channel="nullChannel"
+        				send-timeout="5000" />
+        </integration:poller>
     </inbound-channel-adapter>
 
-	<integration:channel id="resultChannel" />
+	<integration:channel id="successChannel" />
 
     <beans:bean id="filter" class="org.springframework.integration.file.config.FileListFilterFactoryBean"/>
 

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

@@ -30,13 +30,10 @@ 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;
 
@@ -94,15 +91,6 @@ 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> {
 
         public int compare(File f1, File f2) {

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

@@ -218,42 +218,6 @@
 							</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,14 +24,19 @@
 				local-filename-generator-expression="#this.toUpperCase() + '.a'"
 				comparator="comparator"
 				temporary-file-suffix=".foo"
-				remote-directory="foo/bar"
-				disposition-expression="'foo'"
-				disposition-result-channel="dispoChannel"
-				disposition-send-timeout="123">
-			<int:poller fixed-rate="1000"/>
+				remote-directory="foo/bar">
+			<int:poller fixed-rate="1000">
+				<int:psuedo-transactional on-success-expression="'foo'"
+					on-success-result-channel="successChannel"
+					on-failure-expression="'bar'"
+					on-failure-result-channel="failureChannel"
+					send-timeout="123" />
+			</int:poller>
 	</int-ftp:inbound-channel-adapter>
 
-	<int:channel id="dispoChannel" />
+	<int:channel id="successChannel" />
+
+	<int:channel id="failureChannel" />
 	
 	<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

@@ -35,7 +35,6 @@ 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;
@@ -78,12 +77,16 @@ 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("foo", TestUtils.getPropertyValue(adapter, "onSuccessExpression", Expression.class).getValue());
+		assertSame(ac.getBean("successChannel"), TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "defaultChannel"));
 		assertEquals(123L, TestUtils.getPropertyValue(
-				TestUtils.getPropertyValue(source, "dispositionMessagingTemplate"), "sendTimeout"));
+				TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "sendTimeout"));
+		assertEquals("bar", TestUtils.getPropertyValue(adapter, "onFailureExpression", Expression.class).getValue());
+		assertSame(ac.getBean("failureChannel"), TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "defaultChannel"));
+		assertEquals(123L, TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "sendTimeout"));
 	}
 
 	@Test

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

@@ -9,16 +9,15 @@
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
 
-	<int:inbound-channel-adapter channel="queue">
+	<int:inbound-channel-adapter channel="input">
 		<bean class="org.springframework.integration.jdbc.PseudoTransactionalMessageSourceTests$MessageSource"/>
 		<int:poller fixed-delay="2000">
 			<int:transactional />
+			<int:transaction-synchronization/>
 		</int:poller>
 	</int:inbound-channel-adapter>
 
-	<int:channel id="queue">
-		<int:queue />
-	</int:channel>
+	<int:channel id="input" />
 
 	<jdbc:embedded-database id="dataSource" type="HSQL" />
 

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

@@ -22,8 +22,12 @@ import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.PseudoTransactionalMessageSource;
+import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -45,27 +49,39 @@ public class PseudoTransactionalMessageSourceTests {
 
 	private static boolean rolledBack;
 
-	private static boolean doRollback;
+	@Autowired
+	private SubscribableChannel input;
 
 	@Test
 	public void testCommit() throws Exception {
+		MessageHandler handler = new MessageHandler() {
+
+			public void handleMessage(Message<?> message) throws MessagingException {
+			}
+		};
+		input.subscribe(handler);
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
 		assertTrue(committed);
+		input.unsubscribe(handler);
 	}
 
 	@Test
 	public void testRollback() throws Exception {
-		doRollback = true;
+		MessageHandler handler = new MessageHandler() {
+
+			public void handleMessage(Message<?> message) throws MessagingException {
+				throw new RuntimeException("expected");
+			}
+		};
+		input.subscribe(handler);
 		assertTrue(latch2.await(10, TimeUnit.SECONDS));
 		assertTrue(rolledBack);
+		input.unsubscribe(handler);
 	}
 
 	public static class MessageSource implements PseudoTransactionalMessageSource<String, Object> {
 
 		public Message<String> receive() {
-			if (doRollback) {
-				throw new RuntimeException("Expected");
-			}
 			return new GenericMessage<String>("foo");
 		}
 

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

@@ -81,25 +81,25 @@ public class MailReceivingMessageSource implements PseudoTransactionalMessageSou
 		return this.mailReceiver.getTransactionContext();
 	}
 
-	public void afterCommit(MailReceiverContext context) {
+	public void afterCommit(Object context) {
 		Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
-		this.mailReceiver.closeContextAfterSuccess(context);
+		this.mailReceiver.closeContextAfterSuccess((MailReceiverContext) context);
 	}
 
-	public void afterRollback(MailReceiverContext context) {
+	public void afterRollback(Object context) {
 		Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
-		this.mailReceiver.closeContextAfterFailure(context);
+		this.mailReceiver.closeContextAfterFailure((MailReceiverContext) context);
 	}
 
 	/**
-	 * For backwards-compatibility; the mail adapter updates the status before the send.
+	 * For backwards-compatibility; with no tx, 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.
+	 * For backwards-compatibility; with no tx, the mail adapter updates the status before the send.
 	 */
 	public void afterSendNoTx(MailReceiverContext resource) {
 		// No op

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

@@ -221,42 +221,6 @@
 							</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,19 +48,24 @@
 			local-filename-generator-expression="#this.toUpperCase() + '.a'"
 			temporary-file-suffix=".bar"
 			comparator="comparator"
-			delete-remote-files="${delete.remote.files}"
-			disposition-expression="'foo'"
-			disposition-result-channel="dispoChannel"
-			disposition-send-timeout="123">
-		<poller fixed-rate="1000"/>
+			delete-remote-files="${delete.remote.files}">
+		<poller fixed-rate="1000">
+			<psuedo-transactional on-success-expression="'foo'"
+				on-success-result-channel="successChannel"
+				on-failure-expression="'bar'"
+				on-failure-result-channel="failureChannel"
+				send-timeout="123" />
+		</poller>
 	</sftp:inbound-channel-adapter>
 
-	<channel id="dispoChannel" />
-	
+	<channel id="successChannel" />
+
+	<channel id="failureChannel" />
+
 	<beans:bean id="comparator" class="org.mockito.Mockito" factory-method="mock">
 		<beans:constructor-arg value="java.util.Comparator"/>
 	</beans:bean>
-	
+
 	<sftp:inbound-channel-adapter id="sftpAdapter"
 			channel="requestChannel"
 			session-factory="sftpSessionFactory"
@@ -71,7 +76,7 @@
 			delete-remote-files="false">
 		<poller fixed-rate="1000"/>
 	</sftp:inbound-channel-adapter>
-	
+
 	<sftp:inbound-channel-adapter id="sftpAdapterWithPattern"
 			session-factory="sftpSessionFactory"
 			channel="requestChannel"
@@ -82,7 +87,7 @@
 			delete-remote-files="false">
 		<poller fixed-rate="1000"/>
 	</sftp:inbound-channel-adapter>
-	
+
 	<sftp:inbound-channel-adapter id="sftpAdapterNoLocalDir"
 			session-factory="sftpSessionFactory"
 			channel="requestChannel"
@@ -93,7 +98,7 @@
 			delete-remote-files="false">
 		<poller fixed-rate="1000"/>
 	</sftp:inbound-channel-adapter>
-	
+
 	<beans:bean id="filter" class="org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter">
 		<beans:constructor-arg value="."/>
 	</beans:bean>

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

@@ -37,7 +37,6 @@ 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;
@@ -84,12 +83,16 @@ 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("foo", TestUtils.getPropertyValue(adapter, "onSuccessExpression", Expression.class).getValue());
+		assertSame(context.getBean("successChannel"), TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "defaultChannel"));
 		assertEquals(123L, TestUtils.getPropertyValue(
-				TestUtils.getPropertyValue(fileSource, "dispositionMessagingTemplate"), "sendTimeout"));
+				TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "sendTimeout"));
+		assertEquals("bar", TestUtils.getPropertyValue(adapter, "onFailureExpression", Expression.class).getValue());
+		assertSame(context.getBean("failureChannel"), TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "defaultChannel"));
+		assertEquals(123L, TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "sendTimeout"));
 	}
 
 	@Test