INT-1849/INT-2606 Pseudo Transactional Message Src

Initial commit.

Tested with POP3 and IMAP (James) with Sample app.

Essentially moved all the flagging and deleting code
from receive() to closeContextAfterSuccess().

For non-transactional cases, this new method is
called immediately after receiving the message(s),
essentially working as before.

When run from a <transactional/> poller it is
called using TransactionSynchronization after
the transaction commits.

This behavior can be changed by setting
'symchronized="false"' on the poller, which
removes the synchronization and the update
is called immediately after the receive().

Polishing

PR Comments

Update Reference

spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.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.
@@ -33,9 +33,10 @@ import org.springframework.util.Assert;
 
 /**
  * FactoryBean for creating a SourcePollingChannelAdapter instance.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<SourcePollingChannelAdapter>,
 		BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle {
@@ -47,7 +48,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 	private volatile PollerMetadata pollerMetadata;
 
 	private volatile boolean autoStartup = true;
-	
+
 	private volatile Long sendTimeout;
 
 	private volatile String beanName;
@@ -65,7 +66,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 	public void setSource(MessageSource<?> source) {
 		this.source = source;
 	}
-	
+
 	public void setSendTimeout(long sendTimeout) {
 		this.sendTimeout = sendTimeout;
 	}
@@ -138,11 +139,12 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 			spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
 			if (this.sendTimeout != null){
 				spca.setSendTimeout(this.sendTimeout);
-			}		
+			}
 			spca.setTaskExecutor(this.pollerMetadata.getTaskExecutor());
 			spca.setAdviceChain(this.pollerMetadata.getAdviceChain());
 			spca.setTrigger(this.pollerMetadata.getTrigger());
 			spca.setErrorHandler(this.pollerMetadata.getErrorHandler());
+			spca.setSynchronized(this.pollerMetadata.isSynchronized());
 			spca.setBeanClassLoader(this.beanClassLoader);
 			spca.setAutoStartup(this.autoStartup);
 			spca.setBeanName(this.beanName);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,18 +16,19 @@
 
 package org.springframework.integration.config.xml;
 
-import org.w3c.dom.Element;
-
 import org.springframework.beans.BeanMetadataElement;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
 import org.springframework.util.xml.DomUtils;
+import org.w3c.dom.Element;
 
 /**
  * Base parser for inbound Channel Adapters that poll a source.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public abstract class AbstractPollingInboundChannelAdapterParser extends AbstractChannelAdapterParser {
 
@@ -37,8 +38,8 @@ public abstract class AbstractPollingInboundChannelAdapterParser extends Abstrac
 		if (source == null) {
 			parserContext.getReaderContext().error("failed to parse source", element);
 		}
-		BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-				IntegrationNamespaceUtils.BASE_PACKAGE + ".config.SourcePollingChannelAdapterFactoryBean");
+		BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(SourcePollingChannelAdapterFactoryBean.class);
 		adapterBuilder.addPropertyValue("source", source);
 		adapterBuilder.addPropertyReference("outputChannel", channelName);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "send-timeout");

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

@@ -19,10 +19,6 @@ package org.springframework.integration.config.xml;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.w3c.dom.Element;
-import org.w3c.dom.Node;
-import org.w3c.dom.NodeList;
-
 import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.BeanDefinitionHolder;
@@ -34,13 +30,16 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
+import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.scheduling.support.CronTrigger;
 import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
 
 /**
  * Parser for the <poller> element.
@@ -101,6 +100,7 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 			errorHandler.addPropertyReference("defaultErrorChannel", errorChannel);
 			metadataBuilder.addPropertyValue("errorHandler", errorHandler.getBeanDefinition());
 		}
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(metadataBuilder, element, "synchronized");
 		return metadataBuilder.getBeanDefinition();
 	}
 

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

@@ -0,0 +1,64 @@
+/*
+ * 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.core;
+
+import org.springframework.transaction.support.TransactionSynchronization;
+
+/**
+ * {@link MessageSource}s implementing this sub-interface can participate in
+ * a Spring transaction. While the underlying resource is not strictly
+ * transactional, the final disposition of the resource will be
+ * synchronized with any encompassing transaction. For example, when
+ * a message source is used with a transactional poller, if any upstream
+ * activity causes the transaction to roll back, then the {@link #afterRollback(Object)}
+ * method will be called, allowing the message source to reset the state of
+ * whatever. If the transaction commits, the {@link #afterCommit(Object)} method
+ * is called.<p/>
+ * For example, with a MailReceivingMessageSource, the email can be deleted
+ * on successful commit, but not deleted if the transaction rolls back.
+ * <p/>
+ * This implements the 'Best Chance 1PC' pattern where there is only a
+ * 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.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public interface PseudoTransactionalMessageSource<T> extends MessageSource<T> {
+
+	/**
+	 * Obtain the resource on which appropriate action needs
+	 * to be taken.
+	 * @return The resource.
+	 */
+	Object getResource();
+
+	/**
+	 * Invoked via {@link TransactionSynchronization} when the
+	 * transaction commits.
+	 * @param resource The resource to be "committed"
+	 */
+	void afterCommit(Object resource);
+
+	/**
+	 * Invoked via {@link TransactionSynchronization} when the
+	 * transaction rolls back.
+	 * @param resource
+	 */
+	void afterRollback(Object resource);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.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.
@@ -21,32 +21,43 @@ import org.springframework.integration.MessageChannel;
 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.transaction.support.ResourceHolder;
+import org.springframework.transaction.support.ResourceHolderSynchronization;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 
 /**
  * A Channel Adapter implementation for connecting a
  * {@link MessageSource} to a {@link MessageChannel}.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent {
 
 	private volatile MessageSource<?> source;
 
+	private volatile boolean isPseudoTxMessageSource;
+
 	private volatile MessageChannel outputChannel;
 
 	private volatile boolean shouldTrack;
 
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
+	private volatile boolean synchronizedTx = true;
+
 	/**
 	 * Specify the source to be polled for Messages.
 	 */
 	public void setSource(MessageSource<?> source) {
 		this.source = source;
+		this.isPseudoTxMessageSource = this.source instanceof PseudoTransactionalMessageSource;
 	}
 
 	/**
@@ -71,6 +82,10 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 		this.shouldTrack = shouldTrack;
 	}
 
+	public void setSynchronized(boolean synchronizedTx) {
+		this.synchronizedTx = synchronizedTx;
+	}
+
 	@Override
 	public String getComponentType() {
 		return (this.source instanceof NamedComponent) ?
@@ -83,10 +98,43 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 		Assert.notNull(this.outputChannel, "outputChannel must not be null");
 		super.onInit();
 	}
-	
+
 	@Override
 	protected boolean doPoll() {
-		Message<?> message = this.source.receive();
+		boolean isInTx = false;
+		PseudoTransactionalMessageSource<?> messageSource = null;
+		Object resource = null;
+		if (this.isPseudoTxMessageSource) {
+			messageSource = (PseudoTransactionalMessageSource<?>) 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();
+		}
+		finally {
+			if (this.isPseudoTxMessageSource && !isInTx) {
+				/*
+				 * If the message source implements PseudoTransactionalMessageSource and
+				 * we're running from a transactional poller, the message source's afterCommit
+				 * method will be called by the transaction interceptor, using the transaction
+				 * synchronization callback, after the transaction is committed.
+				 *
+				 * If we are not running in a transaction, we invoke it manually, so the message
+				 * source can take the appropriate action, immediately after the receive;
+				 * this was the behavior before pseudo transaction support was added.
+				 */
+				messageSource.afterCommit(resource);
+			}
+		}
 		if (this.logger.isDebugEnabled()){
 			this.logger.debug("Poll resulted in Message: " + message);
 		}
@@ -102,4 +150,67 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
 		}
 		return false;
 	}
+
+	private class PseudoTransactionalResourceHolder implements ResourceHolder {
+
+		private final Object resource;
+
+		public PseudoTransactionalResourceHolder(Object resource) {
+			this.resource = resource;
+		}
+
+		protected Object getResource() {
+			return resource;
+		}
+
+		public void reset() {
+		}
+
+		public void unbound() {
+		}
+
+		public boolean isVoid() {
+			return false;
+		}
+
+	}
+
+	private class PseudoTransactionalResourceSynchronization
+		extends ResourceHolderSynchronization<PseudoTransactionalResourceHolder, Object> {
+
+		private final PseudoTransactionalResourceHolder resourceHolder;
+
+		public PseudoTransactionalResourceSynchronization(PseudoTransactionalResourceHolder resourceHolder,
+				Object resourceKey) {
+			super(resourceHolder, resourceKey);
+			this.resourceHolder = resourceHolder;
+		}
+
+		@Override
+		protected boolean shouldReleaseBeforeCompletion() {
+			return false;
+		}
+
+		@Override
+		protected void processResourceAfterCommit(PseudoTransactionalResourceHolder resourceHolder) {
+			if (logger.isTraceEnabled()) {
+				logger.trace("'Committing' pseudo-transactional resource");
+			}
+			((PseudoTransactionalMessageSource<?>) source).afterCommit(resourceHolder.getResource());
+		}
+
+		@Override
+		public void afterCompletion(int status) {
+			if (status != TransactionSynchronization.STATUS_COMMITTED) {
+				if (logger.isTraceEnabled()) {
+					logger.trace("'Rolling back' pseudo-transactional resource");
+				}
+				((PseudoTransactionalMessageSource<?>) source).afterRollback(this.resourceHolder.getResource());
+			}
+			super.afterCompletion(status);
+		}
+
+
+
+	}
 }

spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java

@@ -20,7 +20,6 @@ import java.util.List;
 import java.util.concurrent.Executor;
 
 import org.aopalliance.aop.Advice;
-
 import org.springframework.scheduling.Trigger;
 import org.springframework.util.ErrorHandler;
 
@@ -31,19 +30,21 @@ import org.springframework.util.ErrorHandler;
 public class PollerMetadata {
 
 	public static final int MAX_MESSAGES_UNBOUNDED = Integer.MIN_VALUE;
-	
+
 	private volatile Trigger trigger;
 
 	private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
 
 	private volatile long receiveTimeout = 1000;
-	
+
 	private volatile ErrorHandler errorHandler;
 
 	private List<Advice> adviceChain;
 
 	private volatile Executor taskExecutor;
 
+	private volatile boolean synchronizedTx = true;
+
 	public void setTrigger(Trigger trigger) {
 		this.trigger = trigger;
 	}
@@ -51,7 +52,7 @@ public class PollerMetadata {
 	public Trigger getTrigger() {
 		return this.trigger;
 	}
-	
+
 	public ErrorHandler getErrorHandler() {
 		return errorHandler;
 	}
@@ -64,9 +65,9 @@ public class PollerMetadata {
 	 * Set the maximum number of messages to receive for each poll.
 	 * A non-positive value indicates that polling should repeat as long
 	 * as non-null messages are being received and successfully sent.
-	 * 
+	 *
 	 * <p>The default is unbounded.
-	 * 
+	 *
 	 * @see #MAX_MESSAGES_UNBOUNDED
 	 */
 	public void setMaxMessagesPerPoll(long maxMessagesPerPoll) {
@@ -100,4 +101,12 @@ public class PollerMetadata {
 	public Executor getTaskExecutor() {
 		return this.taskExecutor;
 	}
+
+	public boolean isSynchronized() {
+		return synchronizedTx;
+	}
+
+	public void setSynchronized(boolean synchronizedTx) {
+		this.synchronizedTx = synchronizedTx;
+	}
 }

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

@@ -1541,6 +1541,20 @@
 				]]></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">

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

@@ -18,16 +18,14 @@ package org.springframework.integration.config.xml;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 import java.util.HashMap;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Test;
-
 import org.aopalliance.aop.Advice;
-
+import org.junit.Test;
 import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -95,6 +93,7 @@ public class PollerParserTests {
 		assertEquals(TransactionInterceptor.class, txAdvice.getClass());
 		TransactionAttributeSource transactionAttributeSource = ((TransactionInterceptor) txAdvice).getTransactionAttributeSource();
 		assertEquals(NameMatchTransactionAttributeSource.class, transactionAttributeSource.getClass());
+		@SuppressWarnings("rawtypes")
 		HashMap nameMap = TestUtils.getPropertyValue(transactionAttributeSource, "nameMap", HashMap.class);
 		assertEquals(1, nameMap.size());
 		assertEquals("{*=PROPAGATION_REQUIRES_NEW,ISOLATION_DEFAULT,readOnly}", nameMap.toString());
@@ -122,13 +121,13 @@ public class PollerParserTests {
 		PollerMetadata metadata = (PollerMetadata) poller;
 		assertTrue(metadata.getTrigger() instanceof TestTrigger);
 	}
-    
+
     @Test(expected=BeanDefinitionParsingException.class)
 	public void pollerWithCronTriggerAndTimeUnit() {
 		new ClassPathXmlApplicationContext(
 				"cronTriggerWithTimeUnit-fail.xml", PollerParserTests.class);
 	}
-    
+
     @Test(expected=BeanDefinitionParsingException.class)
 	public void topLevelPollerWithRef() {
 		new ClassPathXmlApplicationContext(
@@ -141,4 +140,24 @@ 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/config/xml/pollerWithSynchronization.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<poller id="noSync" fixed-delay="3000"/>
+
+	<poller id="syncTrue" fixed-delay="3000" synchronized="true" />
+
+	<poller id="syncFalse" fixed-delay="3000" synchronized="false" />
+
+</beans:beans>

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

@@ -0,0 +1,112 @@
+/*
+ * 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.endpoint;
+
+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.integration.Message;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.PseudoTransactionalMessageSource;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionSynchronizationUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class PseudoTransactionalMessageSourceTests {
+
+	@Test
+	public void testCommit() {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		final Object object = new Object();
+		final AtomicReference<Object> committed = new AtomicReference<Object>();
+		final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
+		adapter.setSource(new PseudoTransactionalMessageSource<String>() {
+
+			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);
+			}
+		});
+
+		TransactionSynchronizationManager.initSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(true);
+		adapter.doPoll();
+		TransactionSynchronizationUtils.triggerAfterCommit();
+		assertSame(object, committed.get());
+		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+		TransactionSynchronizationManager.clearSynchronization();
+		assertNull(rolledBack.get());
+	}
+
+	@Test
+	public void testRollback() {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		final Object object = new Object();
+		final AtomicReference<Object> committed = new AtomicReference<Object>();
+		final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
+		adapter.setSource(new PseudoTransactionalMessageSource<String>() {
+
+			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);
+			}
+		});
+
+		TransactionSynchronizationManager.initSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(true);
+		adapter.doPoll();
+		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+		assertSame(object, rolledBack.get());
+		TransactionSynchronizationManager.clearSynchronization();
+		assertNull(committed.get());
+	}
+
+}