INT-2685 Transaction Synchronization

Remove M3 disposition-* attributes on File/(S)FTP inbound adapters.

Add <pseudo-transactional/> and <transaction-synchronization/> elements
to <poller/>.

These elements provide the following attributes:

* on-success-expression
* on-success-result-channel
* on-failure-expression
* on-failure-result-channel
* send-timeout

<transaction-synchronization/> synchronizes these expression evaluations
with the transaction, such that they are executed immediately after
the commit/rollback.

<psuedo-transactional/> is used for a non-transactional poller.

When an <advice-chain/> is provided to the poller, <psuedo-transactional/>
and <transaction-synchronization/> are synonyms; and the behavior is
dictated by whether or not the <advice-chain/> contains a transaction
advice. It is recommended that <pseudo-transactional/> is used when the
<advice-chain/> does not have a txAdvice, and <transaction-synchronization/>
when it does, but the framework does not enforce this.

The expressions have the original (polled) message as the #root variable.
In addition, a BeanResolver is provided, allowing expressions such as
'@someBean.handleSuccess(payload)'.

MessageSources may also implement PseudoTransactionalMessageSource. This
has a number of methods allowing more flexibility in transactional and
non-transactional environents. For example, for backwards compatibility.
the mail-inbound-channel-adapter deletes its polled message after the
receive() rather than after the polled message is sent (when running in
a non-transactional poller). However, when running in a transactional
poller, the delete is done after the transaction commits (but not when
it rolls back).

In addition, MessageSources that implement this interface can optionally
provide an arbitrary object to the success/failure expressions in a
variable named '#resource'.

INT-2685 Polishing

PR Review Comments

Add tests for non-tx PseudoTransactionalMessageSource

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";
+		}
+	}
 }