From 9bc9867d317f0e1b50edf13e3fbed8e2840099ad Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 12 Jun 2012 13:31:21 -0400 Subject: [PATCH] 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 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 --- ...ourcePollingChannelAdapterFactoryBean.java | 12 +- ...actPollingInboundChannelAdapterParser.java | 13 +- .../integration/config/xml/PollerParser.java | 10 +- .../PseudoTransactionalMessageSource.java | 64 +++++ .../endpoint/SourcePollingChannelAdapter.java | 119 +++++++- .../scheduling/PollerMetadata.java | 21 +- .../config/xml/spring-integration-2.2.xsd | 14 + .../config/xml/PollerParserTests.java | 31 +- .../config/xml/pollerWithSynchronization.xml | 16 ++ ...PseudoTransactionalMessageSourceTests.java | 112 ++++++++ ...ransactionalMessageSourceTests-context.xml | 28 ++ ...PseudoTransactionalMessageSourceTests.java | 87 ++++++ .../mail/AbstractMailReceiver.java | 267 +++++++++++------- .../mail/ImapIdleChannelAdapter.java | 31 +- .../integration/mail/ImapMailReceiver.java | 6 +- .../integration/mail/MailReceiver.java | 41 ++- .../mail/MailReceivingMessageSource.java | 25 +- .../integration/mail/SearchTermStrategy.java | 2 +- .../MailInboundChannelAdapterParser.java | 28 +- .../mail/config/MailReceiverFactoryBean.java | 19 +- .../mail/ImapMailReceiverTests.java | 66 +++-- .../mail/ImapMailSearchTermsTests.java | 42 ++- .../mail/MailReceivingMessageSourceTests.java | 14 +- .../integration/mail/MailTestsHelper.java | 18 +- .../mail/Pop3MailReceiverTests.java | 150 +++++++--- src/reference/docbook/transactions.xml | 15 +- src/reference/docbook/whats-new.xml | 10 +- 27 files changed, 998 insertions(+), 263 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/core/PseudoTransactionalMessageSource.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/pollerWithSynchronization.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests-context.xml create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index 9435800762..ee5521dfc7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/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, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle { @@ -47,7 +48,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean source) { this.source = source; } - + public void setSendTimeout(long sendTimeout) { this.sendTimeout = sendTimeout; } @@ -138,11 +139,12 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean + * For example, with a MailReceivingMessageSource, the email can be deleted + * on successful commit, but not deleted if the transaction rolls back. + *

+ * 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 extends MessageSource { + + /** + * 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); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 31f0424f3d..458f91e4b4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/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 { + + 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); + } + + + + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java index affb5fedf1..176b00bb82 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java +++ b/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 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. - * + * *

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; + } } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd index c25810f846..5c3ce54be6 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd @@ -1541,6 +1541,20 @@ ]]> + + + + 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. + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java index c2ed4c1725..1192367fbc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java +++ b/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()); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/pollerWithSynchronization.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/pollerWithSynchronization.xml new file mode 100644 index 0000000000..8ee7ca9381 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/pollerWithSynchronization.xml @@ -0,0 +1,16 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java new file mode 100644 index 0000000000..9b8b020e79 --- /dev/null +++ b/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 committed = new AtomicReference(); + final AtomicReference rolledBack = new AtomicReference(); + adapter.setSource(new PseudoTransactionalMessageSource() { + + public Message receive() { + return new GenericMessage("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 committed = new AtomicReference(); + final AtomicReference rolledBack = new AtomicReference(); + adapter.setSource(new PseudoTransactionalMessageSource() { + + public Message receive() { + return new GenericMessage("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()); + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests-context.xml new file mode 100644 index 0000000000..8aff133da4 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests-context.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java new file mode 100644 index 0000000000..aa8decad62 --- /dev/null +++ b/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 { + + public Message receive() { + if (doRollback) { + throw new RuntimeException("Expected"); + } + return new GenericMessage("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(); + } + + } +} diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java index 52d4c297df..781419a5b2 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java +++ b/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 contextHolder = new ThreadLocal(); 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 copiedMessages = new LinkedList(); + 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 copiedMessages = new LinkedList(); - 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); + } + } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java index f7150bdcf5..fae74bc06d 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java +++ b/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; } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java index d8f24d146b..7518129a77 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java +++ b/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; } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceiver.java index cb349f0912..161ff0b898 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceiver.java +++ b/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; + } + + } } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java index 27a6c2ecca..cffd5db545 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java +++ b/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 { +public class MailReceivingMessageSource implements PseudoTransactionalMessageSource { private final Log logger = LogFactory.getLog(this.getClass()); @@ -75,4 +77,17 @@ public class MailReceivingMessageSource implements MessageSource, Dispo private volatile Authenticator authenticator; + private volatile boolean synchronizedTx; + /** * Indicates whether retrieved messages should be deleted from the server. * This value will be null unless 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, 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, 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); diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java index 8a3792eb8e..fcbda40c6f 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java +++ b/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 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() { public Object answer(InvocationOnMock invocation) throws Throwable { @@ -371,6 +367,7 @@ public class ImapMailReceiverTests { doAnswer(new Answer() { 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")); } + } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java index 0031219c50..b94c9dc629 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java +++ b/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(); diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/MailReceivingMessageSourceTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/MailReceivingMessageSourceTests.java index e3802c7e00..8397b5ca7e 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/MailReceivingMessageSourceTests.java +++ b/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) { + } + } } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/MailTestsHelper.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/MailTestsHelper.java index b519f27887..0817d45ac4 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/MailTestsHelper.java +++ b/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 contextHolder = TestUtils.getPropertyValue(receiver, "contextHolder",ThreadLocal.class); + Folder folder = mock(Folder.class); + MailReceiverContext context = new MailReceiverContext(folder); + contextHolder.set(context); + return context; + } + } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java index cf62e56d91..6938b3841d 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java +++ b/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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(receiver).openFolder(); + adapter.setSource(new MailReceivingMessageSource(receiver)); + + TransactionSynchronizationManager.initSynchronization(); + TransactionSynchronizationManager.setActualTransactionActive(true); + final AtomicReference doPollMethod = new AtomicReference(); + 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() { + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(receiver).openFolder(); + adapter.setSource(new MailReceivingMessageSource(receiver)); + + TransactionSynchronizationManager.initSynchronization(); + TransactionSynchronizationManager.setActualTransactionActive(true); + final AtomicReference doPollMethod = new AtomicReference(); + 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); + } + } diff --git a/src/reference/docbook/transactions.xml b/src/reference/docbook/transactions.xml index 382d26fa49..0fa15248a6 100644 --- a/src/reference/docbook/transactions.xml +++ b/src/reference/docbook/transactions.xml @@ -150,7 +150,7 @@ ]]> -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 <advice-chain> defined by the Poller. If you only need to address transactional concerns of the Poller, then you can still use the <transactional> 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. + +
+ Pollers and Transaction Synchronization + + 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. + + + 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. + +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 18525c8e4d..4e22bfb60a 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -56,7 +56,15 @@ Stored Procedure Name Expression JdbcCallOperations Cache Statistics - + +
+ Transaction Synchronization + + When running from a transactional poller, + mail inbound adapters can be configured to update the mailbox only + if the transaction commits. + +