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 <transactional/> 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.
This commit is contained in:
Oleg Zhurakousky
@@ -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){
|
||||
|
||||
@@ -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<MessageSourceResourceHolder, Object> {
|
||||
extends ResourceHolderSynchronization<IntegrationResourceHolder, Object> {
|
||||
|
||||
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);
|
||||
|
||||
@@ -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<String, Object> 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 {
|
||||
|
||||
@@ -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<String, Object> attributes = new HashMap<String, Object>();
|
||||
|
||||
public MessageSourceResourceHolder(MessageSource<?> source) {
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
protected MessageSource<?> getMessageSource() {
|
||||
return this.source;
|
||||
}
|
||||
|
||||
public void setMessage(Message<?> message) {
|
||||
this.message = message;
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
}
|
||||
@@ -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<String> receive() {
|
||||
GenericMessage<String> message = new GenericMessage<String>("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<String> receive() {
|
||||
GenericMessage<String> message = new GenericMessage<String>("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<String> receive() {
|
||||
GenericMessage<String> message = new GenericMessage<String>("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<String> receive() {
|
||||
GenericMessage<String> message = new GenericMessage<String>("foo");
|
||||
((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this))
|
||||
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
|
||||
.addAttribute("baz", "qux");
|
||||
return message;
|
||||
}
|
||||
@@ -236,7 +237,7 @@ public class PseudoTransactionalMessageSourceTests {
|
||||
|
||||
public Message<String> receive() {
|
||||
GenericMessage<String> message = new GenericMessage<String>("foo");
|
||||
((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this))
|
||||
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
|
||||
.addAttribute("baz", "qux");
|
||||
return message;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user