From ee91a6ce5aa20558133c68e86b7986487fcf58a5 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 7 Sep 2012 10:00:47 -0400 Subject: [PATCH] INT-1819 Mail pseudo-tx support Added pseudo-tx support for Mail inbound adapters For polling adapters no changes have been made other then returning a javax.mail.Message instead of its copy so post-tx dispositions could be performed on it. For Imap IDLE adapter changes are simiar to the once present in SPCA where TX synchronization logic was added to ImapIdleChannelAdapter Couple of things to note: First IDLE receives an array of messages while Polling task receives one message which means i need to sendMessage in the Polling task in the separate thread, so for maintaining single thread semantics we have now it uses single thread executor to send Messages Renamed MSRH to TransactionalResourceHolder since we no longer use 'source' anywhere and in the case of IDLE there is no MessageSource. Its is truly a holder of attributes we want to make available for use (e.g., SpEL) INT-1819 Polishing - Change TransactionalResourceHolder to IntegrationResourceHolder - Make messageSource available as an attribute - Allow configuration of Executor for ImapIdle adapter - Add parser test for TX ImapIdle adapter - Fix bundlor config for mail - Remove top level element that was added to core - Restore 'legacy' mail attributes in TX, and add schema doc INT-1819 Mail TX Reference Docs Add reference documentation for mail transaction support. INT-1819 Remove 'public abstract' from interface Modifiers are not needed on an interface. --- build.gradle | 3 + .../endpoint/SourcePollingChannelAdapter.java | 7 +- ...aultTransactionSynchronizationFactory.java | 21 +-- ...ngTransactionSynchronizationProcessor.java | 14 +- ...er.java => IntegrationResourceHolder.java} | 13 +- .../TransactionSynchronizationProcessor.java | 6 +- ...PseudoTransactionalMessageSourceTests.java | 17 +- .../mail/AbstractMailReceiver.java | 147 ++++++++++-------- .../mail/ImapIdleChannelAdapter.java | 100 ++++++++++-- .../config/ImapIdleChannelAdapterParser.java | 16 +- .../config/spring-integration-mail-2.2.xsd | 27 +++- .../mail/config/ImapIdelIntegrationTests.java | 98 ++++++++++++ ...pIdleChannelAdapterParserTests-context.xml | 15 ++ .../ImapIdleChannelAdapterParserTests.java | 24 ++- .../imap-idle-mock-integration-config.xml | 32 ++++ .../mongodb/inbound/MongoDbMessageSource.java | 6 +- .../inbound/RedisStoreMessageSource.java | 8 + src/reference/docbook/mail.xml | 61 ++++++++ 18 files changed, 487 insertions(+), 128 deletions(-) rename spring-integration-core/src/main/java/org/springframework/integration/transaction/{MessageSourceResourceHolder.java => IntegrationResourceHolder.java} (85%) create mode 100644 spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdelIntegrationTests.java create mode 100644 spring-integration-mail/src/test/java/org/springframework/integration/mail/config/imap-idle-mock-integration-config.xml diff --git a/build.gradle b/build.gradle index e6a7423004..3e9a7817e9 100644 --- a/build.gradle +++ b/build.gradle @@ -575,14 +575,17 @@ project('spring-integration-mail') { bundleSymbolicName = 'org.springframework.integration.mail' importTemplate += [ 'org.springframework.integration.*;version="[2.2.0, 2.2.1)"', + 'org.springframework.aop.*;version="[3.1.1, 4.0.0)"', 'org.springframework.beans.*;version="[3.1.1, 4.0.0)"', 'org.springframework.expression.*;version="[3.1.1, 4.0.0)"', 'org.springframework.scheduling.*;version="[3.1.1, 4.0.0)"', 'org.springframework.context;version="[3.1.1, 4.0.0)"', 'org.springframework.core.*;version="[3.1.1, 4.0.0)"', 'org.springframework.mail.*;version="[3.1.1, 4.0.0)"', + 'org.springframework.transaction.*;version="[3.1.1, 4.0.0)"', 'org.springframework.util.*;version="[3.1.1, 4.0.0)"', 'org.apache.commons.logging;version="[1.1.1, 2.0.0)"', + 'org.aopalliance.*;version="[1.0.0, 2.0.0)"', 'javax.mail.*;version="[1.4.0, 2.0.0)"', 'com.sun.mail.imap.*;version="[1.4.0, 2.0.0)";resolution:=optional', 'org.w3c.dom.*;version="0"' 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 766d2397cb..f1a594b507 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 @@ -24,7 +24,7 @@ import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.history.TrackableComponent; -import org.springframework.integration.transaction.MessageSourceResourceHolder; +import org.springframework.integration.transaction.IntegrationResourceHolder; import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; @@ -100,10 +100,11 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme protected boolean doPoll() { Message message; - MessageSourceResourceHolder holder = null; + IntegrationResourceHolder holder = null; if (TransactionSynchronizationManager.isActualTransactionActive()) { - holder = new MessageSourceResourceHolder(source); + holder = new IntegrationResourceHolder(); + holder.addAttribute(IntegrationResourceHolder.MESSAGE_SOURCE, source); TransactionSynchronizationManager.bindResource(source, holder); if (transactionSynchronizationFactory != null){ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java index fe949e4755..c121e15135 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java @@ -14,6 +14,7 @@ package org.springframework.integration.transaction; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.transaction.support.ResourceHolderSynchronization; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -41,21 +42,21 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync public TransactionSynchronization create(Object key) { Assert.notNull(key, "'key' must not be null"); Object resourceHolder = TransactionSynchronizationManager.getResource(key); - Assert.isInstanceOf(MessageSourceResourceHolder.class, resourceHolder); - return new DefaultTransactionalResourceSynchronization((MessageSourceResourceHolder) resourceHolder, key); + Assert.isInstanceOf(IntegrationResourceHolder.class, resourceHolder); + return new DefaultTransactionalResourceSynchronization((IntegrationResourceHolder) resourceHolder, key); } /** */ private class DefaultTransactionalResourceSynchronization - extends ResourceHolderSynchronization { + extends ResourceHolderSynchronization { - private final MessageSourceResourceHolder messageSourceHolder; + private final IntegrationResourceHolder resourceHolder; - public DefaultTransactionalResourceSynchronization(MessageSourceResourceHolder messageSourceHolder, + public DefaultTransactionalResourceSynchronization(IntegrationResourceHolder resourceHolder, Object resourceKey) { - super(messageSourceHolder, resourceKey); - this.messageSourceHolder = messageSourceHolder; + super(resourceHolder, resourceKey); + this.resourceHolder = resourceHolder; } @Override @@ -63,7 +64,7 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync if (logger.isTraceEnabled()) { logger.trace("'pre-Committing' transactional resource"); } - processor.processBeforeCommit(messageSourceHolder); + processor.processBeforeCommit(resourceHolder); } @Override @@ -72,7 +73,7 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync } @Override - protected void processResourceAfterCommit(MessageSourceResourceHolder resourceHolder) { + protected void processResourceAfterCommit(IntegrationResourceHolder resourceHolder) { if (logger.isTraceEnabled()) { logger.trace("'Committing' transactional resource"); @@ -89,7 +90,7 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync logger.trace("'Rolling back' transactional resource"); } - processor.processAfterRollback(messageSourceHolder); + processor.processAfterRollback(resourceHolder); } super.afterCompletion(status); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java index 72c3697ad4..4f10bf2c70 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java @@ -87,19 +87,19 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int this.afterRollbackExpression = afterRollbackExpression; } - public void processBeforeCommit(MessageSourceResourceHolder holder) { + public void processBeforeCommit(IntegrationResourceHolder holder) { this.doProcess(holder, this.beforeCommitExpression, this.beforeCommitChannel, "beforeCommit"); } - public void processAfterCommit(MessageSourceResourceHolder holder) { + public void processAfterCommit(IntegrationResourceHolder holder) { this.doProcess(holder, this.afterCommitExpression, this.afterCommitChannel, "afterCommit"); } - public void processAfterRollback(MessageSourceResourceHolder holder) { + public void processAfterRollback(IntegrationResourceHolder holder) { this.doProcess(holder, this.afterRollbackExpression, this.afterRollbackChannel, "afterRollback"); } - private void doProcess(MessageSourceResourceHolder holder, Expression expression, MessageChannel messageChannel, String expressionType) { + private void doProcess(IntegrationResourceHolder holder, Expression expression, MessageChannel messageChannel, String expressionType) { Message message = holder.getMessage(); if (message != null){ if (expression != null){ @@ -161,14 +161,12 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int StandardEvaluationContext evaluationContextToUse; if (resource != null) { evaluationContextToUse = this.createEvaluationContext(); - if (resource instanceof MessageSourceResourceHolder) { - MessageSourceResourceHolder holder = (MessageSourceResourceHolder) resource; + if (resource instanceof IntegrationResourceHolder) { + IntegrationResourceHolder holder = (IntegrationResourceHolder) resource; for (Entry entry : holder.getAttributes().entrySet()) { String key = entry.getKey(); - Assert.state(!("messageSource".equals(key)), "'messageSource' is reserved and cannot be used as an attribute name"); evaluationContextToUse.setVariable(key, entry.getValue()); } - evaluationContextToUse.setVariable("messageSource", holder.getMessageSource()); } } else { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/MessageSourceResourceHolder.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java similarity index 85% rename from spring-integration-core/src/main/java/org/springframework/integration/transaction/MessageSourceResourceHolder.java rename to spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java index d4531e1fc8..989ce07534 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/MessageSourceResourceHolder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java @@ -17,7 +17,6 @@ import java.util.HashMap; import java.util.Map; import org.springframework.integration.Message; -import org.springframework.integration.core.MessageSource; import org.springframework.transaction.support.ResourceHolder; /** @@ -29,22 +28,14 @@ import org.springframework.transaction.support.ResourceHolder; * @since 2.2 * */ -public class MessageSourceResourceHolder implements ResourceHolder { +public class IntegrationResourceHolder implements ResourceHolder { - private final MessageSource source; + public static final String MESSAGE_SOURCE = "messageSource"; private volatile Message message; private final Map attributes = new HashMap(); - public MessageSourceResourceHolder(MessageSource source) { - this.source = source; - } - - protected MessageSource getMessageSource() { - return this.source; - } - public void setMessage(Message message) { this.message = message; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java index 43c999e1ba..868d6001e4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java @@ -22,10 +22,10 @@ package org.springframework.integration.transaction; */ public interface TransactionSynchronizationProcessor { - public abstract void processBeforeCommit(MessageSourceResourceHolder holder); + void processBeforeCommit(IntegrationResourceHolder holder); - public abstract void processAfterCommit(MessageSourceResourceHolder holder); + void processAfterCommit(IntegrationResourceHolder holder); - public abstract void processAfterRollback(MessageSourceResourceHolder holder); + void processAfterRollback(IntegrationResourceHolder holder); } \ No newline at end of file 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 index a7d906cbc6..bc311126f4 100644 --- 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 @@ -19,6 +19,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import org.junit.Test; + import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; @@ -27,8 +28,8 @@ import org.springframework.integration.core.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory; import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor; -import org.springframework.integration.transaction.MessageSourceResourceHolder; import org.springframework.integration.transaction.PseudoTransactionManager; +import org.springframework.integration.transaction.IntegrationResourceHolder; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionSynchronization; @@ -66,8 +67,8 @@ public class PseudoTransactionalMessageSourceTests { public Message receive() { GenericMessage message = new GenericMessage("foo"); - ((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux"); - ((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("bix", "qox"); + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux"); + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("bix", "qox"); return message; } }); @@ -108,7 +109,7 @@ public class PseudoTransactionalMessageSourceTests { public Message receive() { GenericMessage message = new GenericMessage("foo"); - ((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux"); + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux"); return message; } }); @@ -150,8 +151,8 @@ public class PseudoTransactionalMessageSourceTests { public Message receive() { GenericMessage message = new GenericMessage("foo"); - ((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux"); - ((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("bix", "qox"); + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux"); + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("bix", "qox"); return message; } }); @@ -193,7 +194,7 @@ public class PseudoTransactionalMessageSourceTests { public Message receive() { GenericMessage message = new GenericMessage("foo"); - ((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)) + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)) .addAttribute("baz", "qux"); return message; } @@ -236,7 +237,7 @@ public class PseudoTransactionalMessageSourceTests { public Message receive() { GenericMessage message = new GenericMessage("foo"); - ((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)) + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)) .addAttribute("baz", "qux"); return message; } 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..d9a89a1d4b 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 @@ -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; @@ -68,7 +68,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl private volatile Folder folder; private volatile boolean shouldDeleteMessages; - + protected volatile int folderOpenMode = Folder.READ_ONLY; private volatile Properties javaMailProperties = new Properties(); @@ -117,7 +117,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 +129,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 +140,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) { @@ -220,8 +220,8 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl } this.folder.open(this.folderOpenMode); } - - public Message[] receive() throws javax.mail.MessagingException { + + public Message[] receive() throws javax.mail.MessagingException { synchronized (this.folderMonitor) { try { this.openFolder(); @@ -240,68 +240,89 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl 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); + + if (logger.isDebugEnabled()) { + logger.debug("Received " + messages.length + " messages"); } - - 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()]); + + Message[] filteredMessages = this.filterMessagesThruSelector(messages); + + this.postProcessFilteredMessages(filteredMessages); + + return filteredMessages; } finally { MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages); } - } + } + } + + private void postProcessFilteredMessages(Message[] filteredMessages) throws MessagingException { + this.setMessageFlags(filteredMessages); + + if (this.shouldDeleteMessages()) { + this.deleteMessages(filteredMessages); + } + } + + private void setMessageFlags(Message[] filteredMessages) throws MessagingException { + boolean recentFlagSupported = false; + + Flags flags = this.getFolder().getPermanentFlags(); + + if (flags != null){ + recentFlagSupported = flags.contains(Flags.Flag.RECENT); + } + for (Message message : filteredMessages) { + 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); + message.setFlags(siFlags, true); + } + else { + if (logger.isDebugEnabled()){ + logger.debug("USER flags are not supported by this mail server. Flagging message with system flag"); + } + message.setFlag(Flags.Flag.FLAGGED, true); + } + } + this.setAdditionalFlags(message); + } + } + + /** + * Will filter Messages thru selector. Messages that did not pass selector filtering criteria + * will be filtered out and remain on the server as never touched. + */ + private Message[] filterMessagesThruSelector(Message[] messages) throws MessagingException { + List filteredMessages = new LinkedList(); + for (int i = 0; i < messages.length; i++) { + MimeMessage message = (MimeMessage) messages[i]; + if (this.selectorExpression != null) { + if (this.selectorExpression.getValue(this.context, message, Boolean.class)){ + filteredMessages.add(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."); + } + } + } + filteredMessages.add(message); + } + return filteredMessages.toArray(new Message[filteredMessages.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 */ @@ -315,7 +336,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl /** * Deletes the given messages from this receiver's folder. - * + * * @param messages the messages to delete * @throws MessagingException in case of JavaMail errors */ @@ -326,9 +347,9 @@ 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 */ 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..43874d001a 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 @@ -17,20 +17,29 @@ package org.springframework.integration.mail; import java.util.Date; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; import javax.mail.FolderClosedException; import javax.mail.Message; import javax.mail.MessagingException; import javax.mail.Store; -import javax.mail.internet.MimeMessage; +import org.aopalliance.aop.Advice; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.transaction.IntegrationResourceHolder; +import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; +import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; /** * An event-driven Channel Adapter that receives mail messages from a mail @@ -38,17 +47,24 @@ 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 { +public class ImapIdleChannelAdapter extends MessageProducerSupport implements BeanClassLoaderAware { private final IdleTask idleTask = new IdleTask(); + private volatile Executor sendingTaskExecutor = Executors.newFixedThreadPool(1); + private volatile boolean shouldReconnectAutomatically = true; + private volatile ClassLoader classLoader; + + private volatile List adviceChain; + private final ImapMailReceiver mailReceiver; private volatile int reconnectDelay = 10000; // milliseconds @@ -58,15 +74,35 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { private volatile ScheduledFuture pingTask; private volatile long connectionPingInterval = 10000; - + private final ExceptionAwarePeriodicTrigger receivingTaskTrigger = new ExceptionAwarePeriodicTrigger(); + private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; public ImapIdleChannelAdapter(ImapMailReceiver mailReceiver) { Assert.notNull(mailReceiver, "'mailReceiver' must not be null"); this.mailReceiver = mailReceiver; } + public void setTransactionSynchronizationFactory( + TransactionSynchronizationFactory transactionSynchronizationFactory) { + this.transactionSynchronizationFactory = transactionSynchronizationFactory; + } + + public void setAdviceChain(List adviceChain) { + this.adviceChain = adviceChain; + } + + + /** + * Specify an {@link Executor} used to send messages received by the + * adapter. + * @param sendingTaskExecutor the sendingTaskExecutor to set + */ + public void setSendingTaskExecutor(Executor sendingTaskExecutor) { + Assert.notNull(sendingTaskExecutor, "'sendingTaskExecutor' must not be null"); + this.sendingTaskExecutor = sendingTaskExecutor; + } /** * Specify whether the IDLE task should reconnect automatically after @@ -77,10 +113,14 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { this.shouldReconnectAutomatically = shouldReconnectAutomatically; } + @Override public String getComponentType() { return "mail:imap-idle-channel-adapter"; } + public void setBeanClassLoader(ClassLoader classLoader) { + this.classLoader = classLoader; + } /* * Lifecycle implementation @@ -140,9 +180,11 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { if (logger.isDebugEnabled()) { logger.debug("received " + mailMessages.length + " mail messages"); } - for (Message mailMessage : mailMessages) { - final MimeMessage copied = new MimeMessage((MimeMessage) mailMessage); - sendMessage(MessageBuilder.withPayload(copied).build()); + for (final Message mailMessage : mailMessages) { + + Runnable messageSendingTask = createMessageSendingTask(mailMessage); + + sendingTaskExecutor.execute(messageSendingTask); } } } @@ -162,6 +204,39 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { } } + private Runnable createMessageSendingTask(final Message mailMessage){ + Runnable sendingTask = new Runnable() { + public void run() { + org.springframework.integration.Message message = + MessageBuilder.withPayload(mailMessage).build(); + + if (TransactionSynchronizationManager.isActualTransactionActive()) { + + IntegrationResourceHolder holder = new IntegrationResourceHolder(); + holder.setMessage(message); + TransactionSynchronizationManager.bindResource(ImapIdleChannelAdapter.this, holder); + + if (transactionSynchronizationFactory != null){ + TransactionSynchronizationManager. + registerSynchronization(transactionSynchronizationFactory.create(ImapIdleChannelAdapter.this)); + } + } + sendMessage(message); + } + }; + + // wrap in the TX proxy if neccessery + if (!CollectionUtils.isEmpty(adviceChain)) { + ProxyFactory proxyFactory = new ProxyFactory(sendingTask); + if (!CollectionUtils.isEmpty(adviceChain)) { + for (Advice advice : adviceChain) { + proxyFactory.addAdvice(advice); + } + } + sendingTask = (Runnable) proxyFactory.getProxy(classLoader); + } + return sendingTask; + } private class PingTask implements Runnable { @@ -176,9 +251,9 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { } } } - + private class ExceptionAwarePeriodicTrigger implements Trigger { - + private volatile boolean delayNextExecution; @@ -186,15 +261,14 @@ 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/config/ImapIdleChannelAdapterParser.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParser.java index b8d3778edb..359bf8ee4c 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParser.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParser.java @@ -16,8 +16,6 @@ package org.springframework.integration.mail.config; -import org.w3c.dom.Element; - import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; @@ -28,6 +26,8 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.mail.ImapIdleChannelAdapter; import org.springframework.integration.mail.ImapMailReceiver; import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; /** * Parser for the <imap-idle-channel-adapter> element in the 'mail' namespace. @@ -46,7 +46,17 @@ public class ImapIdleChannelAdapterParser extends AbstractChannelAdapterParser { builder.addPropertyReference("outputChannel", channelName); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel", "errorChannel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); - return builder.getBeanDefinition(); + + Element txElement = DomUtils.getChildElementByTagName(element, "transactional"); + if (txElement != null){ + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, txElement, + "synchronization-factory", "transactionSynchronizationFactory"); + } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor", "sendingTaskExecutor"); + AbstractBeanDefinition beanDefinition = builder.getBeanDefinition(); + IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(null, + DomUtils.getChildElementByTagName(element, "transactional"), beanDefinition, parserContext); + return beanDefinition; } private BeanDefinition parseImapMailReceiver(Element element, ParserContext parserContext) { diff --git a/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.2.xsd b/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.2.xsd index ea240f4b1e..7490351374 100644 --- a/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.2.xsd +++ b/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.2.xsd @@ -108,6 +108,9 @@ + + + @@ -140,6 +143,21 @@ + + + + + + + + + + @@ -230,14 +248,19 @@ diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdelIntegrationTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdelIntegrationTests.java new file mode 100644 index 0000000000..cb6f11ea1a --- /dev/null +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdelIntegrationTests.java @@ -0,0 +1,98 @@ +/* + * 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.mail.config; + +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.util.concurrent.atomic.AtomicBoolean; + +import javax.mail.Folder; +import javax.mail.Message; + +import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.mail.ImapIdleChannelAdapter; +import org.springframework.integration.mail.ImapMailReceiver; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.util.ReflectionUtils; +/** + * @author Oleg Zhurakousky + */ +public class ImapIdelIntegrationTests { + + @Test + //@Ignore + public void testWithTransactionSynchronization() throws Exception{ + final AtomicBoolean block = new AtomicBoolean(false); + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("imap-idle-mock-integration-config.xml", this.getClass()); + PostTransactionProcessor processor = context.getBean("syncProcessor", PostTransactionProcessor.class); + ImapIdleChannelAdapter adapter = context.getBean("customAdapter", ImapIdleChannelAdapter.class); + ImapMailReceiver receiver = TestUtils.getPropertyValue(adapter, "mailReceiver", ImapMailReceiver.class); + + // setup mock scenario + receiver = spy(receiver); + + doAnswer(new Answer() { // ensures that waitFornewMessages call blocks after a first execution + // to emulate the behavior of IDLE + + public Object answer(InvocationOnMock invocation) throws Throwable { + if (block.get()){ + Thread.sleep(5000); + } + block.set(true); + return null; + } + }).when(receiver).waitForNewMessages(); + + Message m1 = mock(Message.class); + doReturn(new Message[]{m1}).when(receiver).receive(); + + Folder folder = mock(Folder.class); + when(folder.isOpen()).thenReturn(true); + Field folderField = ReflectionUtils.findField(ImapMailReceiver.class, "folder"); + folderField.setAccessible(true); + folderField.set(receiver, folder); + + Field mrField = ImapIdleChannelAdapter.class.getDeclaredField("mailReceiver"); + mrField.setAccessible(true); + mrField.set(adapter, receiver); + // end mock setup + + adapter.start(); + + Thread.sleep(1000); + // validating that TXpost processor was invoked + + adapter.stop(); + context.destroy(); + verify(processor, Mockito.times(1)).process(m1); + } + + public static interface PostTransactionProcessor { + public void process(Message mailMessage); + } +} diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests-context.xml b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests-context.xml index a8b12caff5..d475a80e48 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests-context.xml +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests-context.xml @@ -66,6 +66,21 @@ should-delete-messages="${mail.delete}" search-term-strategy="searchTermStrategy"/> + + + + + + + + + + diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests.java index 47188a20db..2ab4873b6e 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests.java +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.mail.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; @@ -29,7 +30,6 @@ import javax.mail.search.SearchTerm; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -78,6 +78,7 @@ public class ImapIdleChannelAdapterParserTests { assertEquals(Boolean.TRUE, receiverAccessor.getPropertyValue("shouldDeleteMessages")); assertEquals(Boolean.TRUE, receiverAccessor.getPropertyValue("shouldMarkMessagesAsRead")); assertNull(adapterAccessor.getPropertyValue("errorChannel")); + assertNull(adapterAccessor.getPropertyValue("adviceChain")); } @Test public void simpleAdapterWithErrorChannel() { @@ -161,6 +162,27 @@ public class ImapIdleChannelAdapterParserTests { assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel")); } + @Test + public void transactionalAdapter() { + Object adapter = context.getBean("transactionalAdapter"); + assertEquals(ImapIdleChannelAdapter.class, adapter.getClass()); + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); + Object channel = context.getBean("channel"); + assertSame(channel, adapterAccessor.getPropertyValue("outputChannel")); + assertEquals(Boolean.FALSE, adapterAccessor.getPropertyValue("autoStartup")); + Object receiver = adapterAccessor.getPropertyValue("mailReceiver"); + assertEquals(ImapMailReceiver.class, receiver.getClass()); + DirectFieldAccessor receiverAccessor = new DirectFieldAccessor(receiver); + Object url = receiverAccessor.getPropertyValue("url"); + assertEquals(new URLName("imap:foo"), url); + Properties properties = (Properties) receiverAccessor.getPropertyValue("javaMailProperties"); + assertEquals(0, properties.size()); + assertEquals(Boolean.TRUE, receiverAccessor.getPropertyValue("shouldDeleteMessages")); + assertEquals(Boolean.TRUE, receiverAccessor.getPropertyValue("shouldMarkMessagesAsRead")); + assertNull(adapterAccessor.getPropertyValue("errorChannel")); + assertEquals(context.getBean("executor"), adapterAccessor.getPropertyValue("sendingTaskExecutor")); + assertNotNull(adapterAccessor.getPropertyValue("adviceChain")); + } public static class TestSearchTermStrategy implements SearchTermStrategy { public SearchTerm generateSearchTerm(Flags supportedFlags, Folder folder) { diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/imap-idle-mock-integration-config.xml b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/imap-idle-mock-integration-config.xml new file mode 100644 index 0000000000..7e185089c1 --- /dev/null +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/imap-idle-mock-integration-config.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java index 390a5d2802..0501e397c6 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java @@ -31,7 +31,7 @@ import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; import org.springframework.integration.mongodb.support.MongoHeaders; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.transaction.MessageSourceResourceHolder; +import org.springframework.integration.transaction.IntegrationResourceHolder; import org.springframework.integration.util.ExpressionUtils; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; @@ -218,8 +218,8 @@ public class MongoDbMessageSource extends IntegrationObjectSupport Object holder = TransactionSynchronizationManager.getResource(this); if (holder != null) { - Assert.isInstanceOf(MessageSourceResourceHolder.class, holder); - ((MessageSourceResourceHolder) holder).addAttribute("mongoTemplate", this.mongoTemplate); + Assert.isInstanceOf(IntegrationResourceHolder.class, holder); + ((IntegrationResourceHolder) holder).addAttribute("mongoTemplate", this.mongoTemplate); } return message; diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java index d0c0bec1ad..ebd15e3c89 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java @@ -32,7 +32,9 @@ import org.springframework.integration.Message; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.transaction.IntegrationResourceHolder; import org.springframework.integration.util.ExpressionUtils; +import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; /** * Inbound channel adapter which returns a Message representing a view into @@ -150,6 +152,12 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport RedisStore store = this.createStoreView(key); + Object holder = TransactionSynchronizationManager.getResource(this); + if (holder != null) { + Assert.isInstanceOf(IntegrationResourceHolder.class, holder); + ((IntegrationResourceHolder) holder).addAttribute("store", store); + } + if (store instanceof Collection && ((Collection)store).size() < 1){ return null; } diff --git a/src/reference/docbook/mail.xml b/src/reference/docbook/mail.xml index 9cddae3e84..4ffdda3659 100644 --- a/src/reference/docbook/mail.xml +++ b/src/reference/docbook/mail.xml @@ -207,6 +207,16 @@ In the above example instead of relying on the default SearchTermStra be aware of the protocol you are configuring to retrieve messages. For example POP3 does not support this flag which means setting it to either value will have no effect as messages will NOT be marked as read. + + + It is important to understand that that these actions (marking messages read, and deleting messages) are performed after + the messages are received, but before they are processed. This can cause messages to be lost. + + + You may wish to consider using transaction synchronization instead - see + + + When using the namespace support, a header-enricher Message Transformer is also available. This simplifies the application of the headers mentioned above to any Message prior to sending to the @@ -275,4 +285,55 @@ In the above example instead of relying on the default SearchTermStra +
+ Transaction Synchronization + + Transaction synchronization for inbound adapters allows you to take different actions after a + transaction commits, or rolls back. Transaction synchronization is enabled by adding a <transactional/> element + to the poller for the polled <inbound-adapter/>, or to the <imap-idle-inbound-adapter/>. Even if there + is no 'real' transaction involved, you can still enable this feature by using a + PseudoTransactionManager with the <transactional/> element. + For more information, see . + + + Because of the many different mail servers, and specifically the limitations that some have, at this time we + only provide a strategy for these transaction synchronizations. You can send the messages to some other Spring + Integration components, or invoke a custom bean to perform some action. For example, to move an IMAP message + to a different folder after the transaction commits, you might use something similar to the following: + + + + + + + + + + + +public class Mover { + + public void process(MimeMessage message) throws Exception{ + Folder folder = message.getFolder(); + Store store = null; + if (!folder.isOpen()){ + folder.open(Folder.READ_WRITE); + store = folder.getStore(); + } + message.setFlag(Flags.Flag.DELETED, true); + Folder fooFolder = store.getFolder("FOO")); + fooFolder.appendMessages(new MimeMessage[]{message}); + folder.expunge(); + } +}]]> + + For the message to be still available for manipulation after the transaction, should-delete-messages + must be set to 'false'. + +