Merge pull request #493 from garyrussell/INT-1849

* INT-1849:
  INT-1849/INT-2606 Pseudo Transactional Message Src

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());
+	}
+
+}

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

@@ -0,0 +1,28 @@
+<?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:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:tx="http://www.springframework.org/schema/tx"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		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">
+		<bean class="org.springframework.integration.jdbc.PseudoTransactionalMessageSourceTests$MessageSource"/>
+		<int:poller fixed-delay="2000">
+			<int:transactional />
+		</int:poller>
+	</int:inbound-channel-adapter>
+
+	<int:channel id="queue">
+		<int:queue />
+	</int:channel>
+
+	<jdbc:embedded-database id="dataSource" type="HSQL" />
+
+	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
+		<property name="dataSource" ref="dataSource" />
+	</bean>
+</beans>

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

@@ -0,0 +1,87 @@
+/*
+ * 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.jdbc;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.integration.Message;
+import org.springframework.integration.core.PseudoTransactionalMessageSource;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class PseudoTransactionalMessageSourceTests {
+
+	private static CountDownLatch latch1 = new CountDownLatch(1);
+
+	private static boolean committed;
+
+	private static CountDownLatch latch2 = new CountDownLatch(1);
+
+	private static boolean rolledBack;
+
+	private static boolean doRollback;
+
+	@Test
+	public void testCommit() throws Exception {
+		assertTrue(latch1.await(10, TimeUnit.SECONDS));
+		assertTrue(committed);
+	}
+
+	@Test
+	public void testRollback() throws Exception {
+		doRollback = true;
+		assertTrue(latch2.await(10, TimeUnit.SECONDS));
+		assertTrue(rolledBack);
+	}
+
+	public static class MessageSource implements PseudoTransactionalMessageSource<String> {
+
+		public Message<String> receive() {
+			if (doRollback) {
+				throw new RuntimeException("Expected");
+			}
+			return new GenericMessage<String>("foo");
+		}
+
+		public Object getResource() {
+			return new Object();
+		}
+
+		public void afterCommit(Object resource) {
+			committed = true;
+			latch1.countDown();
+		}
+
+		public void afterRollback(Object resource) {
+			rolledBack = true;
+			latch2.countDown();
+		}
+
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -33,7 +33,6 @@ import javax.mail.internet.MimeMessage;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
@@ -42,17 +41,18 @@ import org.springframework.util.Assert;
 
 /**
  * Base class for {@link MailReceiver} implementations.
- * 
+ *
  * @author Arjen Poutsma
  * @author Jonas Partner
  * @author Mark Fisher
  * @author Iwein Fuld
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public abstract class AbstractMailReceiver extends IntegrationObjectSupport implements MailReceiver, DisposableBean{
 
 	public final static String SI_USER_FLAG = "spring-integration-mail-adapter";
-	
+
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
 	private final URLName url;
@@ -65,10 +65,10 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 
 	private volatile Store store;
 
-	private volatile Folder folder;
+	private final ThreadLocal<MailReceiverContext> contextHolder = new ThreadLocal<MailReceiverContext>();
 
 	private volatile boolean shouldDeleteMessages;
-	
+
 	protected volatile int folderOpenMode = Folder.READ_ONLY;
 
 	private volatile Properties javaMailProperties = new Properties();
@@ -81,8 +81,6 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 
 	protected volatile boolean initialized;
 
-	private final Object folderMonitor = new Object();
-
 
 	public AbstractMailReceiver() {
 		this.url = null;
@@ -117,7 +115,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 	/**
 	 * Set the {@link Session}. Otherwise, the Session will be created by invocation of
 	 * {@link Session#getInstance(Properties)} or {@link Session#getInstance(Properties, Authenticator)}.
-	 * 
+	 *
 	 * @see #setJavaMailProperties(Properties)
 	 * @see #setJavaMailAuthenticator(Authenticator)
 	 */
@@ -129,7 +127,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 	/**
 	 * A new {@link Session} will be created with these properties (and the JavaMailAuthenticator if provided).
 	 * Use either this method or {@link #setSession}, but not both.
-	 * 
+	 *
 	 * @see #setJavaMailAuthenticator(Authenticator)
 	 * @see #setSession(Session)
 	 */
@@ -140,7 +138,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 	/**
 	 * Optional, sets the Authenticator to be used to obtain a session. This will not be used if
 	 * {@link AbstractMailReceiver#setSession} has been used to configure the {@link Session} directly.
-	 * 
+	 *
 	 * @see #setSession(Session)
 	 */
 	public void setJavaMailAuthenticator(Authenticator javaMailAuthenticator) {
@@ -168,7 +166,26 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 	}
 
 	protected Folder getFolder() {
-		return this.folder;
+		return this.getTransactionContext().getFolder();
+	}
+
+	public MailReceiverContext getTransactionContext() {
+		return doObtainTransactionContext();
+	}
+
+	private MailReceiverContext doObtainTransactionContext() {
+		MailReceiverContext mailReceiverContext = this.contextHolder.get();
+		if (mailReceiverContext == null ||
+				mailReceiverContext.getFolder() == null ||
+				!mailReceiverContext.getFolder().isOpen()) {
+			try {
+				this.openFolder();
+			}
+			catch (MessagingException e) {
+				throw new org.springframework.integration.MessagingException("Failed to open folder", e);
+			}
+		}
+		return mailReceiverContext;
 	}
 
 	/**
@@ -204,104 +221,71 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 		}
 	}
 
-	protected void openFolder() throws MessagingException {
+	protected synchronized void openFolder() throws MessagingException {
 		this.openSession();
-		if (this.folder == null) {
-			this.folder = this.store.getFolder(this.url);
+		MailReceiverContext context = this.contextHolder.get();
+		Folder folder = null;
+		if (context == null) {
+			folder = this.store.getFolder(this.url);
+			this.contextHolder.set(new MailReceiverContext(folder));
 		}
-		if (this.folder == null || !this.folder.exists()) {
+		else {
+			folder = context.getFolder();
+		}
+		if (folder == null || !folder.exists()) {
 			throw new IllegalStateException("no such folder [" + this.url.getFile() + "]");
 		}
-		if (this.folder.isOpen()) {
+		if (folder.isOpen()) {
 			return;
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("opening folder [" + MailTransportUtils.toPasswordProtectedString(this.url) + "]");
 		}
-		this.folder.open(this.folderOpenMode);
+		folder.open(this.folderOpenMode);
 	}
-	
-	public Message[] receive() throws javax.mail.MessagingException {	
-		synchronized (this.folderMonitor) {
-			try {
-				this.openFolder();
-				if (logger.isInfoEnabled()) {
-					logger.info("attempting to receive mail from folder [" + this.getFolder().getFullName() + "]");
+
+	public Message[] receive() throws javax.mail.MessagingException {
+		this.openFolder();
+		if (logger.isInfoEnabled()) {
+			logger.info("attempting to receive mail from folder [" + this.getFolder().getFullName() + "]");
+		}
+		Message[] messages = this.searchForNewMessages();
+		if (this.maxFetchSize > 0 && messages.length > this.maxFetchSize) {
+			Message[] reducedMessages = new Message[this.maxFetchSize];
+			System.arraycopy(messages, 0, reducedMessages, 0, this.maxFetchSize);
+			messages = reducedMessages;
+		}
+		if (logger.isDebugEnabled()) {
+			logger.debug("found " + messages.length + " new messages");
+		}
+		if (messages.length > 0) {
+			this.fetchMessages(messages);
+		}
+		List<Message> copiedMessages = new LinkedList<Message>();
+		logger.debug("Recieved " + messages.length + " messages");
+
+		for (int i = 0; i < messages.length; i++) {
+			if (this.selectorExpression != null) {
+				Message message = messages[i];
+				if (this.selectorExpression.getValue(this.context, message, Boolean.class)){
+					copiedMessages.add(new MimeMessage((MimeMessage) message));
 				}
-				Message[] messages = this.searchForNewMessages();
-				if (this.maxFetchSize > 0 && messages.length > this.maxFetchSize) {
-					Message[] reducedMessages = new Message[this.maxFetchSize];
-					System.arraycopy(messages, 0, reducedMessages, 0, this.maxFetchSize);
-					messages = reducedMessages;
-				}
-				if (logger.isDebugEnabled()) {
-					logger.debug("found " + messages.length + " new messages");
-				}
-				if (messages.length > 0) {
-					this.fetchMessages(messages);
-				}
-				List<Message> copiedMessages = new LinkedList<Message>();
-				logger.debug("Recieved " + messages.length + " messages");
-				
-				boolean recentFlagSupported = false;
-				
-				Flags flags = this.getFolder().getPermanentFlags();
-				
-				if (flags != null){
-					recentFlagSupported = flags.contains(Flags.Flag.RECENT);
-				}
-				
-				for (int i = 0; i < messages.length; i++) {
-					if (!recentFlagSupported){
-						if (flags != null && flags.contains(Flags.Flag.USER)){
-							if (logger.isDebugEnabled()){
-								logger.debug("USER flags are supported by this mail server. Flagging message with '" + SI_USER_FLAG + "' user flag");
-							}		
-							Flags siFlags = new Flags();
-							siFlags.add(SI_USER_FLAG);
-							messages[i].setFlags(siFlags, true); 
-						}
-						else {
-							if (logger.isDebugEnabled()){
-								logger.debug("USER flags are not supported by this mail server. Flagging message with system flag");
-							}	
-							messages[i].setFlag(Flags.Flag.FLAGGED, true); 
-						}
-					}
-					if (this.selectorExpression != null) {
-						Message message = messages[i];
-						if (this.selectorExpression.getValue(this.context, message, Boolean.class)){
-							this.setAdditionalFlags(message);
-							copiedMessages.add(new MimeMessage((MimeMessage) message));
-						}	
-						else {
-							if (logger.isDebugEnabled()){
-								logger.debug("Fetched email with subject '" + message.getSubject() + "' will be discarded by the matching filter" +
-												" and will not be flagged as SEEN.");
-							}
-						}
-					}
-					else {
-						this.setAdditionalFlags(messages[i]);
-						copiedMessages.add(new MimeMessage((MimeMessage) messages[i]));
-					}	
-				}
-				if (this.shouldDeleteMessages()) {
-					this.deleteMessages(messages);
-				}
-				return copiedMessages.toArray(new Message[copiedMessages.size()]);
 			}
-			finally {
-				MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
+			else {
+				copiedMessages.add(new MimeMessage((MimeMessage) messages[i]));
 			}
-		}	
+		}
+		if (messages.length > 0) {
+			this.contextHolder.get().setMessages(messages);
+		}
+		return copiedMessages.toArray(new Message[copiedMessages.size()]);
 	}
 
 	/**
 	 * Fetches the specified messages from this receiver's folder. Default
 	 * implementation {@link Folder#fetch(Message[], FetchProfile) fetches}
 	 * every {@link javax.mail.FetchProfile.Item}.
-	 * 
+	 *
 	 * @param messages the messages to fetch
 	 * @throws MessagingException in case of JavaMail errors
 	 */
@@ -310,12 +294,12 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 		contentsProfile.add(FetchProfile.Item.ENVELOPE);
 		contentsProfile.add(FetchProfile.Item.CONTENT_INFO);
 		contentsProfile.add(FetchProfile.Item.FLAGS);
-		this.folder.fetch(messages, contentsProfile);
+		this.contextHolder.get().getFolder().fetch(messages, contentsProfile);
 	}
 
 	/**
 	 * Deletes the given messages from this receiver's folder.
-	 * 
+	 *
 	 * @param messages the messages to delete
 	 * @throws MessagingException in case of JavaMail errors
 	 */
@@ -326,23 +310,20 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 	}
 
 	/**
-	 * Optional method allowing you to set additional flags. 
+	 * Optional method allowing you to set additional flags.
 	 * Currently only implemented in IMapMailReceiver.
-	 * 
+	 *
 	 * @param message
 	 * @throws MessagingException
 	 */
 	protected void setAdditionalFlags(Message message) throws MessagingException {
 	}
 
-	public void destroy() throws Exception {
-		synchronized (this.folderMonitor) {
-			MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
-			MailTransportUtils.closeService(this.store);
-			this.folder = null;
-			this.store = null;
-			this.initialized = false;
-		}
+	public synchronized void destroy() throws Exception {
+		MailTransportUtils.closeFolder(this.contextHolder.get().getFolder(), this.shouldDeleteMessages);
+		MailTransportUtils.closeService(this.store);
+		this.store = null;
+		this.initialized = false;
 	}
 
 	@Override
@@ -361,4 +342,84 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 		return this.store;
 	}
 
+	/**
+	 * Delete and expunge messages after success.
+	 * @param folder
+	 */
+	public void closeContextAfterSuccess(MailReceiverContext context) {
+		Assert.notNull(context, "Mail Reader Context cannot be null");
+		Message[] messages = this.contextHolder.get().getMessages();
+		Assert.state(messages != null, "No messages in mail receiver context");
+		RuntimeException exceptionToThrow = null;
+		boolean recentFlagSupported = false;
+
+		Flags flags = context.getFolder().getPermanentFlags();
+
+		if (flags != null){
+			recentFlagSupported = flags.contains(Flags.Flag.RECENT);
+		}
+
+		for (int i = 0; i < messages.length; i++) {
+			try {
+				if (!recentFlagSupported){
+					if (flags != null && flags.contains(Flags.Flag.USER)){
+						if (logger.isDebugEnabled()){
+							logger.debug("USER flags are supported by this mail server. Flagging message with '" + SI_USER_FLAG + "' user flag");
+						}
+						Flags siFlags = new Flags();
+						siFlags.add(SI_USER_FLAG);
+						messages[i].setFlags(siFlags, true);
+					}
+					else {
+						if (logger.isDebugEnabled()){
+							logger.debug("USER flags are not supported by this mail server. Flagging message with system flag");
+						}
+						messages[i].setFlag(Flags.Flag.FLAGGED, true);
+					}
+				}
+				if (this.selectorExpression != null) {
+					Message message = messages[i];
+					if (this.selectorExpression.getValue(this.context, message, Boolean.class)){
+						this.setAdditionalFlags(message);
+					}
+					else {
+						if (logger.isDebugEnabled()){
+							logger.debug("Fetched email with subject '" + message.getSubject() + "' will be discarded by the matching filter" +
+											" and will not be flagged as SEEN.");
+						}
+					}
+				}
+				else {
+					this.setAdditionalFlags(messages[i]);
+				}
+			}
+			catch (Exception e) {
+				exceptionToThrow = new org.springframework.integration.MessagingException("Failed to set flags", e);
+			}
+		}
+
+		if (this.shouldDeleteMessages) {
+			if (messages != null) {
+				try {
+					this.deleteMessages(messages);
+				}
+				catch (MessagingException e) {
+					exceptionToThrow = new org.springframework.integration.MessagingException("Failed to delete messages", e);
+				}
+			}
+		}
+		Folder folder = context.getFolder();
+		MailTransportUtils.closeFolder(folder, this.shouldDeleteMessages);
+		this.contextHolder.set(null);
+		if (exceptionToThrow != null) {
+			throw exceptionToThrow;
+		}
+	}
+
+	public void closeContextAfterFailure(MailReceiverContext context) {
+		Assert.notNull(context, "Mail Reader Context cannot be null");
+		MailTransportUtils.closeFolder(context.getFolder(), false);
+		this.contextHolder.set(null);
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ package org.springframework.integration.mail;
 import java.util.Date;
 import java.util.concurrent.ScheduledFuture;
 
+import javax.mail.Folder;
 import javax.mail.FolderClosedException;
 import javax.mail.Message;
 import javax.mail.MessagingException;
@@ -26,6 +27,7 @@ import javax.mail.Store;
 import javax.mail.internet.MimeMessage;
 
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.mail.MailReceiver.MailReceiverContext;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.Trigger;
@@ -38,10 +40,11 @@ import org.springframework.util.Assert;
  * messages will be converted and sent as Spring Integration Messages to the
  * output channel. The Message payload will be the {@link javax.mail.Message}
  * instance that was received.
- * 
+ *
  * @author Arjen Poutsma
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public class ImapIdleChannelAdapter extends MessageProducerSupport {
 
@@ -58,7 +61,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 	private volatile ScheduledFuture<?> pingTask;
 
 	private volatile long connectionPingInterval = 10000;
-	
+
 	private final ExceptionAwarePeriodicTrigger receivingTaskTrigger = new ExceptionAwarePeriodicTrigger();
 
 
@@ -77,6 +80,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 		this.shouldReconnectAutomatically = shouldReconnectAutomatically;
 	}
 
+	@Override
 	public String getComponentType() {
 		return "mail:imap-idle-channel-adapter";
 	}
@@ -130,12 +134,16 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 		public void run() {
 			final TaskScheduler scheduler =  getTaskScheduler();
 			Assert.notNull(scheduler, "'taskScheduler' must not be null" );
+			MailReceiverContext context = null;
 			try {
 				if (logger.isDebugEnabled()) {
 					logger.debug("waiting for mail");
 				}
 				mailReceiver.waitForNewMessages();
-				if (mailReceiver.getFolder().isOpen()) {
+				context = mailReceiver.getTransactionContext();
+				Assert.state(context != null, "Mail receiver returned a null context");
+				Folder folder = context.getFolder();
+				if (folder.isOpen()) {
 					Message[] mailMessages = mailReceiver.receive();
 					if (logger.isDebugEnabled()) {
 						logger.debug("received " + mailMessages.length + " mail messages");
@@ -159,6 +167,11 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 							"Failure in 'idle' task. Will NOT resubmit.", e);
 				}
 			}
+			finally {
+				if (context != null) {
+					mailReceiver.closeContextAfterSuccess(context);
+				}
+			}
 		}
 	}
 
@@ -176,9 +189,9 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 			}
 		}
 	}
-	
+
 	private class ExceptionAwarePeriodicTrigger implements Trigger {
-		
+
 		private volatile boolean delayNextExecution;
 
 
@@ -186,12 +199,12 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 			if (delayNextExecution){
 				delayNextExecution = false;
 				return new Date(System.currentTimeMillis() + reconnectDelay);
-			} 
+			}
 			else {
 				return new Date(System.currentTimeMillis());
-			}		
+			}
 		}
-		
+
 		public void delayNextExecution() {
 			this.delayNextExecution = true;
 		}

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

@@ -45,6 +45,7 @@ import com.sun.mail.imap.IMAPMessage;
  * @author Arjen Poutsma
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public class ImapMailReceiver extends AbstractMailReceiver {
 
@@ -102,9 +103,10 @@ public class ImapMailReceiver extends AbstractMailReceiver {
 	 */
 	public void waitForNewMessages() throws MessagingException {
 		this.openFolder();
-		Assert.state(this.getFolder() instanceof IMAPFolder,
+		Folder folder = this.getFolder();
+		Assert.state(folder instanceof IMAPFolder,
 				"folder is not an instance of [" + IMAPFolder.class.getName() + "]");
-		IMAPFolder imapFolder = (IMAPFolder) this.getFolder();
+		IMAPFolder imapFolder = (IMAPFolder) folder;
 		if (imapFolder.hasNewMessages()) {
 			return;
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,13 +16,50 @@
 
 package org.springframework.integration.mail;
 
+import javax.mail.Folder;
+import javax.mail.Message;
+
+import org.springframework.util.Assert;
+
+
 /**
  * Strategy interface for receiving mail {@link javax.mail.Message Messages}.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public interface MailReceiver {
 
 	javax.mail.Message[] receive() throws javax.mail.MessagingException;
 
+	MailReceiverContext getTransactionContext();
+
+	void closeContextAfterSuccess(MailReceiverContext context);
+
+	void closeContextAfterFailure(MailReceiverContext context);
+
+	public static class MailReceiverContext {
+
+		private final Folder folder;
+
+		private volatile Message[] messages = new Message[0];
+
+		MailReceiverContext(Folder folder) {
+			this.folder = folder;
+		}
+
+		Message[] getMessages() {
+			return messages;
+		}
+
+		void setMessages(Message[] messages) {
+			Assert.noNullElements(messages, "messages cannot be null");
+			this.messages = messages;
+		}
+
+		Folder getFolder() {
+			return folder;
+		}
+
+	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,10 +22,11 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.core.PseudoTransactionalMessageSource;
+import org.springframework.integration.mail.MailReceiver.MailReceiverContext;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
 
@@ -33,11 +34,12 @@ import org.springframework.util.Assert;
  * {@link MessageSource} implementation that delegates to a
  * {@link MailReceiver} to poll a mailbox. Each poll of the mailbox may
  * return more than one message which will then be stored in a queue.
- * 
- * @author Jonas Partner 
+ *
+ * @author Jonas Partner
  * @author Mark Fisher
+ * @author Gary Russell
  */
-public class MailReceivingMessageSource implements MessageSource<javax.mail.Message> {
+public class MailReceivingMessageSource implements PseudoTransactionalMessageSource<javax.mail.Message> {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
@@ -75,4 +77,17 @@ public class MailReceivingMessageSource implements MessageSource<javax.mail.Mess
 		return null;
 	}
 
+	public Object getResource() {
+		return this.mailReceiver.getTransactionContext();
+	}
+
+	public void afterCommit(Object context) {
+		Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
+		this.mailReceiver.closeContextAfterSuccess((MailReceiverContext) context);
+	}
+
+	public void afterRollback(Object context) {
+		Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
+		this.mailReceiver.closeContextAfterFailure((MailReceiverContext) context);
+	}
 }

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

@@ -35,7 +35,7 @@ public interface SearchTermStrategy {
 	 *
 	 * @param supportedFlags
 	 * @param folder
-	 * @return
+	 * @return The search term
 	 */
 	SearchTerm generateSearchTerm(Flags supportedFlags, Folder folder);
 

spring-integration-mail/src/main/java/org/springframework/integration/mail/config/MailInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,8 +16,6 @@
 
 package org.springframework.integration.mail.config;
 
-import org.w3c.dom.Element;
-
 import org.springframework.beans.BeanMetadataElement;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
@@ -25,32 +23,31 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.mail.MailReceivingMessageSource;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
+import org.w3c.dom.Element;
 
 /**
- * Parser for the <inbound-channel-adapter> element of Spring Integration's 'mail' namespace. 
- * 
+ * Parser for the <inbound-channel-adapter> element of Spring Integration's 'mail' namespace.
+ *
  * @author Jonas Partner
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public class MailInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
 
-	private static final String BASE_PACKAGE = "org.springframework.integration.mail";
-
-
 	@Override
 	protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
-				BASE_PACKAGE + ".MailReceivingMessageSource");
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MailReceivingMessageSource.class);
 		builder.addConstructorArgValue(this.parseMailReceiver(element, parserContext));
 		return builder.getBeanDefinition();
 	}
 
 	private BeanDefinition parseMailReceiver(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder receiverBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-				BASE_PACKAGE + ".config.MailReceiverFactoryBean");
+				MailReceiverFactoryBean.class);
 		Object source = parserContext.extractSource(element);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "store-uri");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "protocol");
@@ -77,6 +74,9 @@ public class MailInboundChannelAdapterParser extends AbstractPollingInboundChann
 				if (StringUtils.hasText(mmpp)) {
 					receiverBuilder.addPropertyValue("maxFetchSize", mmpp);
 				}
+				if (StringUtils.hasText(pollerElement.getAttribute("synchronized"))) {
+					receiverBuilder.addPropertyValue("synchronized", true);
+				}
 			}
 		}
 		receiverBuilder.addPropertyValue("shouldDeleteMessages", element.getAttribute("should-delete-messages"));
@@ -84,16 +84,16 @@ public class MailInboundChannelAdapterParser extends AbstractPollingInboundChann
 		if (StringUtils.hasText(markAsRead)){
 			receiverBuilder.addPropertyValue("shouldMarkMessagesAsRead", markAsRead);
 		}
-		
+
 		String selectorExpression = element.getAttribute("mail-filter-expression");
-		
+
 		RootBeanDefinition expressionDef = null;
 		if (StringUtils.hasText(selectorExpression)){
 			expressionDef = new RootBeanDefinition("org.springframework.integration.config.ExpressionFactoryBean");
 			expressionDef.getConstructorArgumentValues().addGenericArgumentValue(selectorExpression);
 			receiverBuilder.addPropertyValue("selectorExpression", expressionDef);
 		}
-		
+
 		return receiverBuilder.getBeanDefinition();
 	}
 

spring-integration-mail/src/main/java/org/springframework/integration/mail/config/MailReceiverFactoryBean.java

@@ -24,7 +24,6 @@ import javax.mail.URLName;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.expression.Expression;
@@ -56,16 +55,18 @@ public class MailReceiverFactoryBean implements FactoryBean<MailReceiver>, Dispo
 
 	private volatile Authenticator authenticator;
 
+	private volatile boolean synchronizedTx;
+
 	/**
 	 * Indicates whether retrieved messages should be deleted from the server.
 	 * This value will be <code>null</code> <i>unless</i> explicitly configured.
 	 */
 	private volatile Boolean shouldDeleteMessages = null;
-	
+
 	private volatile Boolean shouldMarkMessagesAsRead = null;
 
 	private volatile int maxFetchSize = 1;
-	
+
 	private volatile Expression selectorExpression;
 
 
@@ -104,11 +105,15 @@ public class MailReceiverFactoryBean implements FactoryBean<MailReceiver>, Dispo
 	public void setMaxFetchSize(int maxFetchSize) {
 		this.maxFetchSize = maxFetchSize;
 	}
-	
+
 	public void setSelectorExpression(Expression selectorExpression) {
 		this.selectorExpression = selectorExpression;
 	}
 
+	public void setSynchronized(boolean synchronizedTx) {
+		this.synchronizedTx = synchronizedTx;
+	}
+
 	public MailReceiver getObject() throws Exception {
 		if (this.receiver == null) {
 			this.receiver = this.createReceiver();
@@ -163,6 +168,12 @@ public class MailReceiverFactoryBean implements FactoryBean<MailReceiver>, Dispo
 			// otherwise, the default is true for POP3 but false for IMAP
 			receiver.setShouldDeleteMessages(this.shouldDeleteMessages);
 		}
+		if (this.synchronizedTx && this.maxFetchSize != 1) {
+			if (logger.isWarnEnabled()) {
+				logger.warn("Max Fetch Size is set to 1 because the poller is synchronized; was " + this.maxFetchSize);
+			}
+			this.maxFetchSize = 1;
+		}
 		receiver.setMaxFetchSize(this.maxFetchSize);
 		receiver.setSelectorExpression(selectorExpression);
 

spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@ package org.springframework.integration.mail;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -24,8 +25,9 @@ import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-import java.lang.reflect.Field;
 import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.mail.Flags;
@@ -42,7 +44,6 @@ import org.junit.Test;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
-
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -51,6 +52,7 @@ import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.mail.MailReceiver.MailReceiverContext;
 import org.springframework.integration.mail.config.ImapIdleChannelAdapterParserTests;
 import org.springframework.integration.test.util.TestUtils;
 
@@ -71,11 +73,9 @@ public class ImapMailReceiverTests {
 		((ImapMailReceiver)receiver).setShouldMarkMessagesAsRead(true);
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
 
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
@@ -105,6 +105,7 @@ public class ImapMailReceiverTests {
 			}
 		}).when(receiver).fetchMessages(messages);
 		receiver.receive();
+		receiver.closeContextAfterSuccess(context);
 		verify(msg1, times(1)).setFlag(Flag.SEEN, true);
 		verify(msg2, times(1)).setFlag(Flag.SEEN, true);
 		verify(receiver, times(0)).deleteMessages((Message[]) Mockito.any());
@@ -117,11 +118,9 @@ public class ImapMailReceiverTests {
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
 
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
 
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
@@ -149,6 +148,7 @@ public class ImapMailReceiverTests {
 			}
 		}).when(receiver).fetchMessages(messages);
 		receiver.receive();
+		receiver.closeContextAfterSuccess(context);
 		verify(msg1, times(1)).setFlag(Flag.SEEN, true);
 		verify(msg2, times(1)).setFlag(Flag.SEEN, true);
 		verify(receiver, times(1)).deleteMessages((Message[]) Mockito.any());
@@ -160,11 +160,9 @@ public class ImapMailReceiverTests {
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
 
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
 
 
 		Message msg1 = mock(MimeMessage.class);
@@ -189,6 +187,7 @@ public class ImapMailReceiverTests {
 		}).when(receiver).fetchMessages(messages);
 		receiver.afterPropertiesSet();
 		receiver.receive();
+		receiver.closeContextAfterFailure(context);
 		verify(msg1, times(0)).setFlag(Flag.SEEN, true);
 		verify(msg2, times(0)).setFlag(Flag.SEEN, true);
 	}
@@ -200,11 +199,9 @@ public class ImapMailReceiverTests {
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
 
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
 
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
@@ -233,6 +230,7 @@ public class ImapMailReceiverTests {
 		}).when(receiver).fetchMessages(messages);
 		receiver.afterPropertiesSet();
 		receiver.receive();
+		receiver.closeContextAfterSuccess(context);
 		verify(msg1, times(0)).setFlag(Flag.SEEN, true);
 		verify(msg2, times(0)).setFlag(Flag.SEEN, true);
 		verify(msg1, times(1)).setFlag(Flag.DELETED, true);
@@ -244,11 +242,9 @@ public class ImapMailReceiverTests {
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
 
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
 
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
@@ -276,6 +272,7 @@ public class ImapMailReceiverTests {
 			}
 		}).when(receiver).fetchMessages(messages);
 		receiver.receive();
+		receiver.closeContextAfterSuccess(context);
 		verify(msg1, times(1)).setFlag(Flag.SEEN, true);
 		verify(msg2, times(1)).setFlag(Flag.SEEN, true);
 		verify(receiver, times(0)).deleteMessages((Message[]) Mockito.any());
@@ -357,11 +354,10 @@ public class ImapMailReceiverTests {
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
 
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(IMAPFolder.class);
+		@SuppressWarnings("unchecked")
+		final ThreadLocal<MailReceiverContext> contextHolder = TestUtils.getPropertyValue(receiver, "contextHolder", ThreadLocal.class);
+		final Folder folder = mock(IMAPFolder.class);
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
 
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
@@ -371,6 +367,7 @@ public class ImapMailReceiverTests {
 
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
+				contextHolder.set(new MailReceiverContext(folder));
 				return null;
 			}
 		}).when(receiver).openFolder();
@@ -404,7 +401,12 @@ public class ImapMailReceiverTests {
 	@Test // see INT-1801
 	public void testImapLifecycleForRaceCondition() throws Exception{
 
-		for (int i = 0; i < 1000; i++) {
+		int count = 1000;
+		final CountDownLatch receiveLatch = new CountDownLatch(count);
+		final CountDownLatch destroyLatch = new CountDownLatch(count);
+		final AtomicInteger receiveCount = new AtomicInteger();
+		final AtomicInteger destroyCount = new AtomicInteger();
+		for (int i = 0; i < count; i++) {
 			final ImapMailReceiver receiver = new ImapMailReceiver("imap://foo");
 			Store store = mock(Store.class);
 			Folder folder = mock(Folder.class);
@@ -429,7 +431,8 @@ public class ImapMailReceiverTests {
 							failed.getAndIncrement();
 						}
 					}
-
+					receiveCount.incrementAndGet();
+					receiveLatch.countDown();
 				}
 			}).start();
 
@@ -441,9 +444,13 @@ public class ImapMailReceiverTests {
 						// ignore
 						ignore.printStackTrace();
 					}
+					destroyCount.incrementAndGet();
+					destroyLatch.countDown();
 				}
 			}).start();
 		}
+		assertTrue("Only " + receiveCount.get() + " receive() calls", receiveLatch.await(10, TimeUnit.SECONDS));
+		assertTrue("Only " + receiveCount.get() + " destroy() calls", destroyLatch.await(10, TimeUnit.SECONDS));
 		assertEquals(0, failed.get());
 	}
 
@@ -455,4 +462,5 @@ public class ImapMailReceiverTests {
 		receiver.setSearchTermStrategy(stStrategy);
 		assertEquals(stStrategy, TestUtils.getPropertyValue(receiver, "searchTermStrategy"));
 	}
+
 }

spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,10 +17,8 @@ package org.springframework.integration.mail;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 
 import javax.mail.Flags;
@@ -32,11 +30,12 @@ import javax.mail.search.NotTerm;
 import javax.mail.search.SearchTerm;
 
 import org.junit.Test;
-
+import org.springframework.integration.mail.MailReceiver.MailReceiverContext;
 import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  *
  */
 public class ImapMailSearchTermsTests {
@@ -45,13 +44,12 @@ public class ImapMailSearchTermsTests {
 	public void validateSearchTermsWhenShouldMarkAsReadNoExistingFlags() throws Exception {
 		ImapMailReceiver receiver = new ImapMailReceiver();
 		receiver.setShouldMarkMessagesAsRead(true);
-		
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
+		when(folder.isOpen()).thenReturn(true);
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
-		
+
 		Method compileSearchTerms = ReflectionUtils.findMethod(receiver.getClass(), "compileSearchTerms", Flags.class);
 		compileSearchTerms.setAccessible(true);
 		Flags flags = new Flags();
@@ -66,14 +64,13 @@ public class ImapMailSearchTermsTests {
 	public void validateSearchTermsWhenShouldMarkAsReadWithExistingFlags() throws Exception {
 		ImapMailReceiver receiver = new ImapMailReceiver();
 		receiver.setShouldMarkMessagesAsRead(true);
-		
+
 		receiver.afterPropertiesSet();
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
+		when(folder.isOpen()).thenReturn(true);
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
-		
+
 		Method compileSearchTerms = ReflectionUtils.findMethod(receiver.getClass(), "compileSearchTerms", Flags.class);
 		compileSearchTerms.setAccessible(true);
 		Flags flags = new Flags();
@@ -90,19 +87,18 @@ public class ImapMailSearchTermsTests {
 		siFlags.add(AbstractMailReceiver.SI_USER_FLAG);
 		assertTrue(((FlagTerm)notTerm.getTerm()).getFlags().contains(siFlags));
 	}
-	
+
 	@Test
 	public void validateSearchTermsWhenShouldNotMarkAsReadNoExistingFlags() throws Exception {
 		ImapMailReceiver receiver = new ImapMailReceiver();
 		receiver.setShouldMarkMessagesAsRead(false);
 		receiver.afterPropertiesSet();
-		
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
+		when(folder.isOpen()).thenReturn(true);
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
-		
+
 		Method compileSearchTerms = ReflectionUtils.findMethod(receiver.getClass(), "compileSearchTerms", Flags.class);
 		compileSearchTerms.setAccessible(true);
 		Flags flags = new Flags();

spring-integration-mail/src/test/java/org/springframework/integration/mail/MailReceivingMessageSourceTests.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.
@@ -29,6 +29,7 @@ import org.junit.Test;
 /**
  * @author Jonas Partner
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public class MailReceivingMessageSourceTests {
 
@@ -71,6 +72,17 @@ public class MailReceivingMessageSourceTests {
 
 		public void stop() {
 		}
+
+		public MailReceiverContext getTransactionContext() {
+			return null;
+		}
+
+		public void closeContextAfterSuccess(MailReceiverContext context) {
+		}
+
+		public void closeContextAfterFailure(MailReceiverContext context) {
+		}
+
 	}
 
 }

spring-integration-mail/src/test/java/org/springframework/integration/mail/MailTestsHelper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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,12 +16,19 @@
 
 package org.springframework.integration.mail;
 
+import static org.mockito.Mockito.mock;
+
+import javax.mail.Folder;
+
 import org.springframework.integration.Message;
+import org.springframework.integration.mail.MailReceiver.MailReceiverContext;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.mail.SimpleMailMessage;
 
 /**
  * @author Marius Bogoevici
+ * @author Gary Russell
  */
 public class MailTestsHelper {
 
@@ -66,4 +73,13 @@ public class MailTestsHelper {
 				.build();
 	}
 
+	public static MailReceiverContext setupContextHolder(AbstractMailReceiver receiver) {
+		@SuppressWarnings("unchecked")
+		ThreadLocal<MailReceiverContext> contextHolder = TestUtils.getPropertyValue(receiver, "contextHolder",ThreadLocal.class);
+		Folder folder = mock(Folder.class);
+		MailReceiverContext context = new MailReceiverContext(folder);
+		contextHolder.set(context);
+		return context;
+	}
+
 }

spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.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.
@@ -23,10 +23,11 @@ import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.concurrent.atomic.AtomicReference;
 
-import javax.mail.Flags.Flag;
 import javax.mail.Flags;
+import javax.mail.Flags.Flag;
 import javax.mail.Folder;
 import javax.mail.Message;
 import javax.mail.internet.MimeMessage;
@@ -35,9 +36,18 @@ import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.mail.MailReceiver.MailReceiverContext;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionSynchronizationUtils;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.util.ReflectionUtils.MethodCallback;
 
 /**
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  *
  */
 public class Pop3MailReceiverTests {
@@ -47,13 +57,11 @@ public class Pop3MailReceiverTests {
 		((Pop3MailReceiver)receiver).setShouldDeleteMessages(true);
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
-		
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
-		
+
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
 		final Message[] messages = new Message[]{msg1, msg2};
@@ -67,13 +75,13 @@ public class Pop3MailReceiverTests {
 				return null;
 			}
 		}).when(receiver).openFolder();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return messages;
 			}
 		}).when(receiver).searchForNewMessages();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return null;
@@ -81,6 +89,7 @@ public class Pop3MailReceiverTests {
 		}).when(receiver).fetchMessages(messages);
 		receiver.afterPropertiesSet();
 		receiver.receive();
+		receiver.closeContextAfterSuccess(context);
 		verify(msg1, times(1)).setFlag(Flag.DELETED, true);
 		verify(msg2, times(1)).setFlag(Flag.DELETED, true);
 	}
@@ -90,13 +99,11 @@ public class Pop3MailReceiverTests {
 		((Pop3MailReceiver)receiver).setShouldDeleteMessages(false);
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
-		
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
-		
+
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
 		final Message[] messages = new Message[]{msg1, msg2};
@@ -105,13 +112,13 @@ public class Pop3MailReceiverTests {
 				return null;
 			}
 		}).when(receiver).openFolder();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return messages;
 			}
 		}).when(receiver).searchForNewMessages();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return null;
@@ -119,6 +126,7 @@ public class Pop3MailReceiverTests {
 		}).when(receiver).fetchMessages(messages);
 		receiver.afterPropertiesSet();
 		receiver.receive();
+		receiver.closeContextAfterFailure(context);
 		verify(msg1, times(0)).setFlag(Flag.DELETED, true);
 		verify(msg2, times(0)).setFlag(Flag.DELETED, true);
 	}
@@ -127,13 +135,11 @@ public class Pop3MailReceiverTests {
 		AbstractMailReceiver receiver = new Pop3MailReceiver("pop3://some.host");
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
-		
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
-		
+
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
 		final Message[] messages = new Message[]{msg1, msg2};
@@ -142,13 +148,13 @@ public class Pop3MailReceiverTests {
 				return null;
 			}
 		}).when(receiver).openFolder();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return messages;
 			}
 		}).when(receiver).searchForNewMessages();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return null;
@@ -156,6 +162,7 @@ public class Pop3MailReceiverTests {
 		}).when(receiver).fetchMessages(messages);
 		receiver.afterPropertiesSet();
 		receiver.receive();
+		receiver.closeContextAfterFailure(context);
 		verify(msg1, times(0)).setFlag(Flag.DELETED, true);
 		verify(msg2, times(0)).setFlag(Flag.DELETED, true);
 	}
@@ -164,13 +171,11 @@ public class Pop3MailReceiverTests {
 		AbstractMailReceiver receiver = new Pop3MailReceiver();
 		receiver = spy(receiver);
 		receiver.afterPropertiesSet();
-		
-		Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
-		folderField.setAccessible(true);
-		Folder folder = mock(Folder.class);
+
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
 		when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
-		folderField.set(receiver, folder);
-		
+
 		Message msg1 = mock(MimeMessage.class);
 		Message msg2 = mock(MimeMessage.class);
 		final Message[] messages = new Message[]{msg1, msg2};
@@ -179,13 +184,13 @@ public class Pop3MailReceiverTests {
 				return null;
 			}
 		}).when(receiver).openFolder();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return messages;
 			}
 		}).when(receiver).searchForNewMessages();
-		
+
 		doAnswer(new Answer<Object>() {
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return null;
@@ -193,7 +198,82 @@ public class Pop3MailReceiverTests {
 		}).when(receiver).fetchMessages(messages);
 		receiver.afterPropertiesSet();
 		receiver.receive();
+		receiver.closeContextAfterFailure(context);
 		verify(msg1, times(0)).setFlag(Flag.DELETED, true);
 		verify(msg2, times(0)).setFlag(Flag.DELETED, true);
 	}
+
+	@Test
+	public void testCommit() throws Exception {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		Pop3MailReceiver receiver = new Pop3MailReceiver("pop3://some.host");
+		receiver.setShouldDeleteMessages(true);
+		receiver = spy(receiver);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
+		when(folder.isOpen()).thenReturn(true);
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				return null;
+			}
+		}).when(receiver).openFolder();
+		adapter.setSource(new MailReceivingMessageSource(receiver));
+
+		TransactionSynchronizationManager.initSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(true);
+		final AtomicReference<Method> doPollMethod = new AtomicReference<Method>();
+		ReflectionUtils.doWithMethods(SourcePollingChannelAdapter.class, new MethodCallback() {
+
+			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+				if (method.getName() == "doPoll") {
+					doPollMethod.set(method);
+					method.setAccessible(true);
+				}
+			}
+		});
+		doPollMethod.get().invoke(adapter, (Object[]) null);
+		TransactionSynchronizationUtils.triggerAfterCommit();
+		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+		TransactionSynchronizationManager.clearSynchronization();
+		verify(folder).close(true);
+	}
+
+	@Test
+	public void testRollback() throws Exception {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		Pop3MailReceiver receiver = new Pop3MailReceiver("pop3://some.host");
+		receiver.setShouldDeleteMessages(true);
+		receiver = spy(receiver);
+		MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
+		Folder folder = context.getFolder();
+		when(folder.isOpen()).thenReturn(true);
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				return null;
+			}
+		}).when(receiver).openFolder();
+		adapter.setSource(new MailReceivingMessageSource(receiver));
+
+		TransactionSynchronizationManager.initSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(true);
+		final AtomicReference<Method> doPollMethod = new AtomicReference<Method>();
+		ReflectionUtils.doWithMethods(SourcePollingChannelAdapter.class, new MethodCallback() {
+
+			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+				if (method.getName() == "doPoll") {
+					doPollMethod.set(method);
+					method.setAccessible(true);
+				}
+			}
+		});
+		doPollMethod.get().invoke(adapter, (Object[]) null);
+		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
+		TransactionSynchronizationManager.clearSynchronization();
+		verify(folder).close(false);
+	}
+
 }

src/reference/docbook/transactions.xml

@@ -150,7 +150,7 @@
 </tx:advice>
 ]]></programlisting>
 
-As yo can see from the example above, we have provided a very basic XML-based configuration of Spring Transaction advice  - "txAdvice" and
+As you can see from the example above, we have provided a very basic XML-based configuration of Spring Transaction advice  - "txAdvice" and
 included it within the <emphasis>&lt;advice-chain&gt;</emphasis> defined by the Poller.
 If you only need to address transactional concerns of the Poller, then you can still use the <emphasis>&lt;transactional&gt;</emphasis> element
 as a convinience.
@@ -173,4 +173,17 @@ as a convinience.
       that delegates to a transactional MessageStore strategy, or you could use a JMS-backed channel.
       </para>
     </section>
+
+    <section id="poller-synch">
+      <title>Pollers and Transaction Synchronization</title>
+      <para>
+        Certain inbound adapters are capable of synchronizing their updates with a transaction. For example, the mail
+        inbound adapters, if running in a transaction and configured to mark or delete messages, will only take those
+        actions on a mail message if the transaction commits; otherwise the mail message is left in the inbox.
+      </para>
+      <para>
+        For all message sources that implement PseudoTransactionalMessageSource, this is the default behavior (commit and
+        rollback detection). It can be disabled by setting the synchronized attribute on the poller to false.
+      </para>
+    </section>
 </chapter>

src/reference/docbook/whats-new.xml

@@ -56,7 +56,15 @@
               <listitem>Stored Procedure Name Expression</listitem>
               <listitem>JdbcCallOperations Cache Statistics</listitem>
             </itemizedlist>
-		</section>
+        </section>
+        <section id="2.2-tx">
+            <title>Transaction Synchronization</title>
+            <para>
+              When running from a transactional poller,
+              mail inbound adapters can be configured to update the mailbox only
+              if the transaction commits.
+            </para>
+        </section>
     </section>
 
     <section id="2.2-new-components">