Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2727 PseudoTX Refactoring

Browse Source 
Remove the need for pseudo-transactional element

INT-2727 PseudoTX
Add PseudoTransactionalTransactionManager

INT-2727
addressed PR comments
cherry picked previous code for mail module to eliminate breaking change

INT-2727
initial refactoring pseudo-tx support to use common configuration

INT-2727
finalizing pseudo-tx synchronization support

INT-2727 polishing

INT-2727 polishing based on PR comments

INT-2727 addressed PR comments

INT-2727 polishing

INT-2727 Remove PseudoTransactionalMessageSource

Instead of getResource, bind the resource holder before
receive() and then add attributes to the holder.

INT-2727 polishing

INT-2727 Polishing

Remove bind of #resource; add beforeCommit() test;
add TransactionTemplate tests.
This commit is contained in:
Oleg Zhurakousky
2012-08-27 13:16:21 -04:00
committed by Gary Russell
parent 5828f70321
commit 94cc5a73e2
50 changed files with 1523 additions and 1357 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -618,6 +618,7 @@ project('spring-integration-mongodb') {
'org.springframework.context;version="[3.1.1, 4.0.0)"',
'org.springframework.core.*;version="[3.1.1, 4.0.0)"',
'org.springframework.expression.*;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.springframework.jmx.*;version="[3.1.1, 4.0.0)"',
'org.springframework.data.mongodb.*;version="[1.0.0, 2.0.0)"',
@@ -654,6 +655,7 @@ project('spring-integration-redis') {
'org.springframework.context;version="[3.1.1, 4.0.0)"',
'org.springframework.core.*;version="[3.1.1, 4.0.0)"',
'org.springframework.expression.*;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.springframework.data.redis.*;version="[1.0.0, 2.0.0)"',
'javax.*;version="0"',

2
spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java
View File

@@ -89,8 +89,6 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
public static final String CONTENT_TYPE = "content-type";
public static final String DISPOSITION_RESULT = "dispositionResult";
public static final String POSTPROCESS_RESULT = "postProcessResult";

22
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -24,7 +24,6 @@ import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.expression.Expression;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageSource;
@@ -145,30 +144,11 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
spca.setAdviceChain(this.pollerMetadata.getAdviceChain());
spca.setTrigger(this.pollerMetadata.getTrigger());
spca.setErrorHandler(this.pollerMetadata.getErrorHandler());
Expression onSuccessExpression = this.pollerMetadata.getOnSuccessExpression();
if (onSuccessExpression != null) {
spca.setOnSuccessExpression(onSuccessExpression);
}
MessageChannel onSuccessResultChannel = this.pollerMetadata.getOnSuccessResultChannel();
if (onSuccessResultChannel != null) {
spca.setOnSuccessResultChannel(onSuccessResultChannel);
}
Expression onFailureExpression = this.pollerMetadata.getOnFailureExpression();
if (onFailureExpression != null) {
spca.setOnFailureExpression(onFailureExpression);
}
MessageChannel onFailureResultChannel = this.pollerMetadata.getOnFailureResultChannel();
if (onFailureResultChannel != null) {
spca.setOnFailureChannel(onFailureResultChannel);
}
Long resultSendTimeout = this.pollerMetadata.getSendTimeout();
if (resultSendTimeout != null) {
spca.setResultSendTimeout(resultSendTimeout);
}
spca.setBeanClassLoader(this.beanClassLoader);
spca.setAutoStartup(this.autoStartup);
spca.setBeanName(this.beanName);
spca.setBeanFactory(this.beanFactory);
spca.setTransactionSynchronizationFactory(this.pollerMetadata.getTransactionSynchronizationFactory());
spca.afterPropertiesSet();
this.adapter = spca;
this.initialized = true;

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

@@ -16,19 +16,21 @@
package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Base parser for inbound Channel Adapters that poll a source.
*
* @author Mark Fisher
* @author Gary Russell
* @author Oleg Zhurakousky
*/
public abstract class AbstractPollingInboundChannelAdapterParser extends AbstractChannelAdapterParser {

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
View File

@@ -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.
@@ -23,6 +23,7 @@ package org.springframework.integration.config.xml;
* @author Marius Bogoevici
* @author Oleg Zhurakousky
* @author David Turanski
* @author Gary Russell
*/
public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@@ -72,6 +73,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
registerBeanDefinitionParser("message-history", new MessageHistoryParser());
registerBeanDefinitionParser("control-bus", new ControlBusParser());
registerBeanDefinitionParser("wire-tap", new GlobalWireTapParser());
registerBeanDefinitionParser("transaction-synchronization-factory", new TransactionSynchronizationFactoryParser());
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -201,7 +201,8 @@ public abstract class IntegrationNamespaceUtils {
"A 'poller' element that provides a 'ref' must have no child elements.", pollerElement);
}
targetBuilder.addPropertyReference("pollerMetadata", pollerElement.getAttribute("ref"));
} else {
}
else {
BeanDefinition beanDefinition = parserContext.getDelegate().parseCustomElement(pollerElement,
targetBuilder.getBeanDefinition());
if (beanDefinition == null) {

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

@@ -19,22 +19,21 @@ package org.springframework.integration.config.xml;
import java.util.ArrayList;
import java.util.List;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Parser for the &lt;poller&gt; element.
@@ -93,18 +92,14 @@ public class PollerParser extends AbstractBeanDefinitionParser {
IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(adviceChainElement, txElement,
metadataBuilder, parserContext);
Element pseudoTxElement = DomUtils.getChildElementByTagName(element, "pseudo-transactional");
if (pseudoTxElement != null && txElement != null) {
parserContext.getReaderContext().error(
"Cannot have both 'transactional' and 'pseudo-transactional' elements", element);
if (txElement != null){
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, txElement,
"synchronization-factory", "transactionSynchronizationFactory");
}
Element txSyncElement = DomUtils.getChildElementByTagName(element, "transaction-synchronization");
if (pseudoTxElement != null && txSyncElement != null) {
parserContext.getReaderContext().error(
"Cannot have both 'transaction-synchronization' and 'pseudo-transactional' elements", element);
else if (adviceChainElement != null){
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, adviceChainElement,
"synchronization-factory", "transactionSynchronizationFactory");
}
pseudoTxElement = pseudoTxElement == null ? txSyncElement : pseudoTxElement;
configureTransactionSync(pseudoTxElement, metadataBuilder, parserContext);
String errorChannel = element.getAttribute("error-channel");
if (StringUtils.hasText(errorChannel)) {
@@ -169,25 +164,4 @@ public class PollerParser extends AbstractBeanDefinitionParser {
}
targetBuilder.addPropertyReference("trigger", triggerBeanNames.get(0));
}
private void configureTransactionSync(Element element, BeanDefinitionBuilder metadataBuilder,
ParserContext parserContext) {
if (element != null) {
configureSyncExpression(element, metadataBuilder, parserContext, "on-success-expression", "onSuccessExpression");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "on-success-result-channel");
configureSyncExpression(element, metadataBuilder, parserContext, "on-failure-expression", "onFailureExpression");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "on-failure-result-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(metadataBuilder, element, "send-timeout", "sendTimeout");
}
}
private void configureSyncExpression(Element element, BeanDefinitionBuilder metadataBuilder,
ParserContext parserContext, String expressionAttribute, String expressionProperty) {
String expression = element.getAttribute(expressionAttribute);
if (StringUtils.hasText(expression)) {
RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(expression);
metadataBuilder.addPropertyValue(expressionProperty, expressionDef);
}
}
}

102
spring-integration-core/src/main/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParser.java Normal file
View File

@@ -0,0 +1,102 @@
/*
* 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.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory;
import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Parser for transaction-synchronizatioin-factory element
*
* @author Oleg Zhurakousky
* @since 2.2
*
*/
public class TransactionSynchronizationFactoryParser extends
AbstractBeanDefinitionParser {
@Override
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
BeanDefinitionBuilder syncFactoryBuilder =
BeanDefinitionBuilder.genericBeanDefinition(DefaultTransactionSynchronizationFactory.class);
Element beforeCommitElement = DomUtils.getChildElementByTagName(element, "before-commit");
Element afterCommitElement = DomUtils.getChildElementByTagName(element, "after-commit");
Element afterRollbackElement = DomUtils.getChildElementByTagName(element, "after-rollback");
if (this.elementsNotDefined(beforeCommitElement, afterCommitElement, afterRollbackElement)){
parserContext.getReaderContext().error("At least one sub-element " +
"('before-commit', 'after-commit' and/or 'after-rollback') must be defined", element);
}
BeanDefinitionBuilder expressionProcessor =
BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingTransactionSynchronizationProcessor.class);
this.processSubElement(beforeCommitElement, parserContext, expressionProcessor, "beforeCommit");
this.processSubElement(afterCommitElement, parserContext, expressionProcessor, "afterCommit");
this.processSubElement(afterRollbackElement, parserContext, expressionProcessor, "afterRollback");
syncFactoryBuilder.addConstructorArgValue(expressionProcessor.getBeanDefinition());
return syncFactoryBuilder.getBeanDefinition();
}
private void processSubElement(Element element, ParserContext parserContext, BeanDefinitionBuilder expressionProcessor, String elementPrefix){
if (element != null){
String expression = element.getAttribute("expression");
String channel = element.getAttribute("channel");
if (this.attributesNotDefined(expression, channel)){
parserContext.getReaderContext().error("At least one attribute " +
"('expression' and/or 'channel') must be defined", element);
}
if (StringUtils.hasText(expression)){
RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(expression);
expressionProcessor.addPropertyValue(elementPrefix + "Expression", expressionDef);
}
if (StringUtils.hasText(channel)){
expressionProcessor.addPropertyReference(elementPrefix + "Channel", channel);
}
else {
expressionProcessor.addPropertyReference(elementPrefix + "Channel", "nullChannel");
}
}
}
private boolean elementsNotDefined(Element... elements){
for (Object element : elements) {
if (element != null){
return false;
}
}
return true;
}
private boolean attributesNotDefined(String... attributes){
for (String attribute : attributes) {
if (StringUtils.hasText(attribute)){
return false;
}
}
return true;
}
}

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

@@ -1,83 +0,0 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.core;
import org.springframework.transaction.support.TransactionSynchronization;
/**
* {@link MessageSource}s implementing this sub-interface can participate in
* a Spring transaction. While the underlying resource is not strictly
* transactional, the final disposition of the resource will be
* synchronized with any encompassing transaction. For example, when
* a message source is used with a transactional poller, if any upstream
* activity causes the transaction to roll back, then the {@link #afterRollback(Object)}
* method will be called, allowing the message source to reset the state of
* whatever. If the transaction commits, the {@link #afterCommit(Object)} method
* is called.<p/>
* For example, with a MailReceivingMessageSource, the email can be deleted
* on successful commit, but not deleted if the transaction rolls back.
* <p/>
* This implements the 'Best Chance 1PC' pattern where there is only a
* small (but present) window in which a transaction might commit but the
* resource is not updated to reflect that. This could result in
* duplicate messages.
* <p>All {@link MessageSource}s can have success/failure expressions evaluated either as part
* of a transaction with a &lt;transactional/&gt; poller or after success/failure when
* running in a &lt;pseudo-transactional/&gt; poller. This interface is for those
* message sources that need additional flexibility than that provided by SpEL expressions.
* @author Gary Russell
* @since 2.2
*
*/
public interface PseudoTransactionalMessageSource<T, V> extends MessageSource<T> {
/**
* Obtain the resource on which appropriate action needs
* to be taken. This resource is passed back into the other
* methods. In addition, it is made available to transaction
* synchronization SpEL expressions in the '#resource' variable.
* @return The resource.
*/
V getResource();
/**
* Invoked via {@link TransactionSynchronization} when the
* transaction commits.
* @param object The resource to be "committed"
*/
void afterCommit(Object object);
/**
* Invoked via {@link TransactionSynchronization} when the
* transaction rolls back.
* @param object
*/
void afterRollback(Object object);
/**
* Called when there is no transaction and the receive() call completed.
* @param resource
*/
void afterReceiveNoTx(V resource);
/**
* Called when there is no transaction and after the message was
* sent to the channel.
* @param resource
*/
void afterSendNoTx(V resource);
}

261
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -16,24 +16,16 @@
package org.springframework.integration.endpoint;
import org.springframework.context.expression.BeanFactoryResolver;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.MessagingException;
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.integration.support.MessageBuilder;
import org.springframework.integration.util.ExpressionUtils;
import org.springframework.transaction.support.ResourceHolder;
import org.springframework.transaction.support.ResourceHolderSynchronization;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.integration.transaction.MessageSourceResourceHolder;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
@@ -47,39 +39,26 @@ import org.springframework.util.Assert;
*/
public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent {
/**
* Transaction synchronization needs a non-null resource; this constant is used for
* message sources that have no need for a resource, because the post-process
* action just needs the message.
*/
private static final Object NO_TX_RESOURCE = new Object();
private volatile MessageSource<?> source;
private volatile boolean isPseudoTxMessageSource;
private volatile MessageChannel outputChannel;
private volatile boolean shouldTrack;
private final MessagingTemplate messagingTemplate = new MessagingTemplate();
private volatile Expression onSuccessExpression;
private volatile TransactionSynchronizationFactory transactionSynchronizationFactory;
private final MessagingTemplate onSuccessMessagingTemplate = new MessagingTemplate();
private volatile Expression onFailureExpression;
private final MessagingTemplate onFailureMessagingTemplate = new MessagingTemplate();
private volatile StandardEvaluationContext evaluationContext;
public void setTransactionSynchronizationFactory(
TransactionSynchronizationFactory transactionSynchronizationFactory) {
this.transactionSynchronizationFactory = transactionSynchronizationFactory;
}
/**
* Specify the source to be polled for Messages.
*/
public void setSource(MessageSource<?> source) {
this.source = source;
this.isPseudoTxMessageSource = this.source instanceof PseudoTransactionalMessageSource;
}
/**
@@ -104,32 +83,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
this.shouldTrack = shouldTrack;
}
public void setOnSuccessExpression(Expression onSuccessExpression) {
Assert.notNull(onSuccessExpression, "onSuccessExpression cannot be null");
this.onSuccessExpression = onSuccessExpression;
}
public void setOnFailureExpression(Expression onFailureExpression) {
Assert.notNull(onFailureExpression, "onFailureExpression cannot be null");
this.onFailureExpression = onFailureExpression;
}
public void setOnSuccessResultChannel(MessageChannel onSuccessResultChannel) {
Assert.notNull(onSuccessResultChannel, "onSuccessChannel cannot be null");
this.onSuccessMessagingTemplate.setDefaultChannel(onSuccessResultChannel);
}
public void setOnFailureChannel(MessageChannel onFailureResultChannel) {
Assert.notNull(onFailureResultChannel, "onFailureChannel cannot be null");
this.onFailureMessagingTemplate.setDefaultChannel(onFailureResultChannel);
}
public void setResultSendTimeout(long sendTimeout) {
this.onSuccessMessagingTemplate.setSendTimeout(sendTimeout);
this.onFailureMessagingTemplate.setSendTimeout(sendTimeout);
}
@Override
public String getComponentType() {
return (this.source instanceof NamedComponent) ?
@@ -140,73 +93,39 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
protected void onInit() {
Assert.notNull(this.source, "source must not be null");
Assert.notNull(this.outputChannel, "outputChannel must not be null");
this.evaluationContext = this.createEvaluationContext();
super.onInit();
}
@SuppressWarnings("unchecked")
@Override
protected boolean doPoll() {
boolean isInTx = false;
PseudoTransactionalMessageSource<?,Object> messageSource = null;
Object resource = NO_TX_RESOURCE;
if (this.isPseudoTxMessageSource) {
messageSource = (PseudoTransactionalMessageSource<?,Object>) this.source;
}
Message<?> message;
try {
message = this.source.receive();
if (this.isPseudoTxMessageSource) {
Object actualResource = messageSource.getResource();
resource = actualResource != null ? actualResource : resource;
}
if (TransactionSynchronizationManager.isActualTransactionActive()) {
TransactionSynchronizationManager.bindResource(this, resource);
TransactionSynchronizationManager.registerSynchronization(
new PseudoTransactionalResourceSynchronization(
new PseudoTransactionalResourceHolder(message, resource), this));
isInTx = true;
}
}
finally {
if (!isInTx && this.isPseudoTxMessageSource) {
/*
* This callback is provided for 'legacy' message sources
* that used to take action after the receive and before
* the send. When in a transaction, that action is now
* taken after the commit but, when not in a transaction
* this callback provides backwards compatibility. An
* example is the mail reader that deletes from the inbox.
*/
messageSource.afterReceiveNoTx(resource);
Message<?> message;
MessageSourceResourceHolder holder = null;
if (TransactionSynchronizationManager.isActualTransactionActive()) {
holder = new MessageSourceResourceHolder(source);
TransactionSynchronizationManager.bindResource(source, holder);
if (transactionSynchronizationFactory != null){
TransactionSynchronizationManager.registerSynchronization(transactionSynchronizationFactory.create(source));
}
}
message = this.source.receive();
if (this.logger.isDebugEnabled()){
this.logger.debug("Poll resulted in Message: " + message);
}
if (message != null) {
if (holder != null) {
holder.setMessage(message);
}
if (this.shouldTrack) {
message = MessageHistory.write(message, this);
}
try {
this.messagingTemplate.send(this.outputChannel, message);
if (!isInTx) {
if (this.isPseudoTxMessageSource) {
/*
* For 'legacy' message sources that need more flexibility than simple
* expression evaluation, we invoke this callback after a
* successful send.
*/
messageSource.afterSendNoTx(resource);
}
this.onSuccess(message, resource);
}
}
catch (Exception e) {
if (!isInTx) {
this.onFailure(message, resource);
}
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
@@ -221,142 +140,4 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
}
return false;
}
private void onSuccess(Message<?> message, Object resource) {
doPostProcess(message, resource, this.onSuccessExpression, this.onSuccessMessagingTemplate, "success");
}
private void onFailure(Message<?> message, Object resource) {
doPostProcess(message, resource, this.onFailureExpression, this.onFailureMessagingTemplate, "failure");
}
private void doPostProcess(Message<?> message, Object resource, Expression expression,
MessagingTemplate messagingTemplate, String expressionType) {
if (expression != null && message != null) {
if (logger.isDebugEnabled()) {
logger.debug("Evaluating " + expressionType + " expression: '" + expression.getExpressionString() + "' on " + message);
}
StandardEvaluationContext evaluationContextToUse = this.determineEvaluationContextToUse(resource);
Object value;
try {
value = expression.getValue(evaluationContextToUse, message);
}
catch (Exception e) {
value = e;
}
if (value != null) {
try {
messagingTemplate.send(MessageBuilder.fromMessage(message)
.setHeader(MessageHeaders.DISPOSITION_RESULT, value).build());
}
catch (Exception e) {
logger.error("Failed to send " + expressionType + " evaluation result " + message, e);
}
}
}
}
/**
* If we don't need a resource variable (not a {@link PseudoTransactionalMessageSource})
* we can use a singleton context; otherwise we need a new one each time.
* @param resource The resource
* @return The context.
*/
private StandardEvaluationContext determineEvaluationContextToUse(Object resource) {
StandardEvaluationContext evaluationContextToUse;
if (resource != NO_TX_RESOURCE) {
evaluationContextToUse = this.createEvaluationContext();
evaluationContextToUse.setVariable("resource", resource);
}
else {
if (this.evaluationContext == null) {
this.evaluationContext = this.createEvaluationContext();
}
evaluationContextToUse = this.evaluationContext;
}
return evaluationContextToUse;
}
protected StandardEvaluationContext createEvaluationContext(){
if (this.getBeanFactory() != null) {
return ExpressionUtils.createStandardEvaluationContext(new BeanFactoryResolver(this.getBeanFactory()),
this.getConversionService());
}
else {
return ExpressionUtils.createStandardEvaluationContext(this.getConversionService());
}
}
private class PseudoTransactionalResourceHolder implements ResourceHolder {
private final Message<?> message;
private final Object resource;
public PseudoTransactionalResourceHolder(Message<?> message, Object resource) {
this.message = message;
this.resource = resource;
}
protected Object getResource() {
return resource;
}
public Message<?> getMessage() {
return message;
}
public void reset() {
}
public void unbound() {
}
public boolean isVoid() {
return false;
}
}
private class PseudoTransactionalResourceSynchronization
extends ResourceHolderSynchronization<PseudoTransactionalResourceHolder, Object> {
private final PseudoTransactionalResourceHolder resourceHolder;
public PseudoTransactionalResourceSynchronization(PseudoTransactionalResourceHolder resourceHolder,
Object resourceKey) {
super(resourceHolder, resourceKey);
this.resourceHolder = resourceHolder;
}
@Override
protected boolean shouldReleaseBeforeCompletion() {
return false;
}
@Override
protected void processResourceAfterCommit(PseudoTransactionalResourceHolder resourceHolder) {
if (logger.isTraceEnabled()) {
logger.trace("'Committing' pseudo-transactional resource");
}
if (isPseudoTxMessageSource) {
((PseudoTransactionalMessageSource<?, ?>) source).afterCommit(resourceHolder.getResource());
}
onSuccess(resourceHolder.getMessage(), resourceHolder.getResource());
}
@Override
public void afterCompletion(int status) {
if (status != TransactionSynchronization.STATUS_COMMITTED) {
if (logger.isTraceEnabled()) {
logger.trace("'Rolling back' pseudo-transactional resource");
}
if (isPseudoTxMessageSource) {
((PseudoTransactionalMessageSource<?, ?>) source).afterRollback(resourceHolder.getResource());
}
onFailure(this.resourceHolder.getMessage(), this.resourceHolder.getResource());
}
super.afterCompletion(status);
}
}
}

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

@@ -20,9 +20,10 @@ import java.util.List;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
import org.springframework.expression.Expression;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.scheduling.Trigger;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
/**
@@ -46,16 +47,20 @@ public class PollerMetadata {
private volatile Executor taskExecutor;
private volatile Expression onSuccessExpression;
private volatile MessageChannel onSuccessResultChannel;
private volatile Expression onFailureExpression;
private volatile MessageChannel onFailureResultChannel;
private volatile long sendTimeout;
private volatile TransactionSynchronizationFactory transactionSynchronizationFactory;
public void setTransactionSynchronizationFactory(
TransactionSynchronizationFactory transactionSynchronizationFactory) {
Assert.notNull(transactionSynchronizationFactory, "'transactionSynchronizationFactory' must not be null");
this.transactionSynchronizationFactory = transactionSynchronizationFactory;
}
public TransactionSynchronizationFactory getTransactionSynchronizationFactory() {
return transactionSynchronizationFactory;
}
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
}
@@ -113,38 +118,6 @@ public class PollerMetadata {
return this.taskExecutor;
}
public Expression getOnSuccessExpression() {
return onSuccessExpression;
}
public void setOnSuccessExpression(Expression onSuccessExpression) {
this.onSuccessExpression = onSuccessExpression;
}
public MessageChannel getOnSuccessResultChannel() {
return onSuccessResultChannel;
}
public void setOnSuccessResultChannel(MessageChannel onSuccessResultChannel) {
this.onSuccessResultChannel = onSuccessResultChannel;
}
public Expression getOnFailureExpression() {
return onFailureExpression;
}
public void setOnFailureExpression(Expression onFailureExpression) {
this.onFailureExpression = onFailureExpression;
}
public MessageChannel getOnFailureResultChannel() {
return onFailureResultChannel;
}
public void setOnFailureResultChannel(MessageChannel onFailureResultChannel) {
this.onFailureResultChannel = onFailureResultChannel;
}
public long getSendTimeout() {
return sendTimeout;
}

99
spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java Normal file
View File

@@ -0,0 +1,99 @@
/*
* 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.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;
import org.springframework.util.Assert;
/**
* Default implementation of {@link TransactionSynchronizationFactory} which takes an instance of
* {@link TransactionSynchronizationProcessor} allowing you to create a {@link TransactionSynchronization}
* using {{@link #create(Object)} method.
*
* @author Gary Russell
* @author Oleg Zhurakousky
* @since 2.2
*/
public class DefaultTransactionSynchronizationFactory implements TransactionSynchronizationFactory {
private final Log logger = LogFactory.getLog(getClass());
private final TransactionSynchronizationProcessor processor;
public DefaultTransactionSynchronizationFactory(TransactionSynchronizationProcessor processor){
Assert.notNull(processor, "'processor' must not be null");
this.processor = processor;
}
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);
}
/**
*/
private class DefaultTransactionalResourceSynchronization
extends ResourceHolderSynchronization<MessageSourceResourceHolder, Object> {
private final MessageSourceResourceHolder messageSourceHolder;
public DefaultTransactionalResourceSynchronization(MessageSourceResourceHolder messageSourceHolder,
Object resourceKey) {
super(messageSourceHolder, resourceKey);
this.messageSourceHolder = messageSourceHolder;
}
@Override
public void beforeCommit(boolean readOnly) {
if (logger.isTraceEnabled()) {
logger.trace("'pre-Committing' transactional resource");
}
processor.processBeforeCommit(messageSourceHolder);
}
@Override
protected boolean shouldReleaseBeforeCompletion() {
return false;
}
@Override
protected void processResourceAfterCommit(MessageSourceResourceHolder resourceHolder) {
if (logger.isTraceEnabled()) {
logger.trace("'Committing' transactional resource");
}
processor.processAfterCommit(resourceHolder);
}
@Override
public void afterCompletion(int status) {
if (status != TransactionSynchronization.STATUS_COMMITTED) {
if (logger.isTraceEnabled()) {
logger.trace("'Rolling back' transactional resource");
}
processor.processAfterRollback(messageSourceHolder);
}
super.afterCompletion(status);
}
}
}

192
spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java Normal file
View File

@@ -0,0 +1,192 @@
/*
* 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.transaction;
import java.util.Map.Entry;
import org.springframework.context.expression.BeanFactoryResolver;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.ExpressionUtils;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.util.Assert;
/**
* This implementation of {@link TransactionSynchronizationFactory}
* allows you to configure SpEL expressions, with their execution being coordinated (synchronized) with a
* transaction - see {@link TransactionSynchronization}. Expressions for before-commit, after-commit, and after-rollback
* are supported, together with a channel for each where the evaluation result (if any) will be sent.
* For each sub-element you can specify 'expression' and/or 'channel' attributes.
* If only the 'channel' attribute is present the received Message will be sent there as part of a particular synchronization scenario.
* If only the 'expression' attribute is present and the result of an expression is a non-Null value, a Message with the
* result as the payload will be generated and sent to a default channel (NullChannel) and will appear in the logs.
* If you want the evaluation result to go to a specific channel add a 'channel' attribute. If the result of an expression is null
* or void, no Message will be generated.
*
* @author Gary Russell
* @author Oleg Zhurakousky
* @since 2.2
*
*/
public class ExpressionEvaluatingTransactionSynchronizationProcessor extends IntegrationObjectSupport implements TransactionSynchronizationProcessor {
private volatile StandardEvaluationContext evaluationContext;
private volatile Expression beforeCommitExpression;
private volatile Expression afterCommitExpression;
private volatile Expression afterRollbackExpression;
private volatile MessageChannel beforeCommitChannel;
private volatile MessageChannel afterCommitChannel;
private volatile MessageChannel afterRollbackChannel;
public void setBeforeCommitChannel(MessageChannel beforeCommitChannel) {
Assert.notNull(beforeCommitChannel, "'beforeCommitChannel' must not be null");
this.beforeCommitChannel = beforeCommitChannel;
}
public void setAfterCommitChannel(MessageChannel afterCommitChannel) {
Assert.notNull(afterCommitChannel, "'afterCommitChannel' must not be null");
this.afterCommitChannel = afterCommitChannel;
}
public void setAfterRollbackChannel(MessageChannel afterRollbackChannel) {
Assert.notNull(afterRollbackChannel, "'afterRollbackChannel' must not be null");
this.afterRollbackChannel = afterRollbackChannel;
}
public void setBeforeCommitExpression(Expression beforeCommitExpression) {
Assert.notNull(beforeCommitExpression, "'beforeCommitExpression' must not be null");
this.beforeCommitExpression = beforeCommitExpression;
}
public void setAfterCommitExpression(Expression afterCommitExpression) {
Assert.notNull(afterCommitExpression, "'afterCommitExpression' must not be null");
this.afterCommitExpression = afterCommitExpression;
}
public void setAfterRollbackExpression(Expression afterRollbackExpression) {
Assert.notNull(afterRollbackExpression, "'afterRollbackExpression' must not be null");
this.afterRollbackExpression = afterRollbackExpression;
}
public void processBeforeCommit(MessageSourceResourceHolder holder) {
this.doProcess(holder, this.beforeCommitExpression, this.beforeCommitChannel, "beforeCommit");
}
public void processAfterCommit(MessageSourceResourceHolder holder) {
this.doProcess(holder, this.afterCommitExpression, this.afterCommitChannel, "afterCommit");
}
public void processAfterRollback(MessageSourceResourceHolder holder) {
this.doProcess(holder, this.afterRollbackExpression, this.afterRollbackChannel, "afterRollback");
}
private void doProcess(MessageSourceResourceHolder holder, Expression expression, MessageChannel messageChannel, String expressionType) {
Message<?> message = holder.getMessage();
if (message != null){
if (expression != null){
if (logger.isDebugEnabled()) {
logger.debug("Evaluating " + expressionType + " expression: '" + expression.getExpressionString() + "' on " + message);
}
StandardEvaluationContext evaluationContextToUse = this.prepareEvaluationContextToUse(holder);
Object value = expression.getValue(evaluationContextToUse, message);
if (value != null) {
Message<?> spelResultMessage = null;
if (logger.isDebugEnabled()) {
logger.debug("Sending expression result message to " + messageChannel + " " +
"as part of '" + expressionType + "' transaction synchronization");
}
try {
spelResultMessage = MessageBuilder.withPayload(value).build();
this.sendMessage(messageChannel, spelResultMessage);
}
catch (Exception e) {
logger.error("Failed to send " + expressionType + " evaluation result " + spelResultMessage, e);
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace("Expression evaluation returned null");
}
}
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Sending received message to " + messageChannel + " as part of '" +
expressionType + "' transaction synchronization");
}
try {
// rollback will be initiated if any of the previous sync operations fail (e.g., beforeCommit)
// this means that this method will be called without explicit configuration thus no channel
if (messageChannel != null){
this.sendMessage(messageChannel, MessageBuilder.fromMessage(message).build());
}
} catch (Exception e) {
logger.error("Failed to send " + message, e);
}
}
}
}
private void sendMessage(MessageChannel channel, Message<?> message){
channel.send(message, 0);
}
/**
* If we don't need variables (i.e., resource is null)
* we can use a singleton context; otherwise we need a new one each time.
* @param resource The resource
* @return The context.
*/
private StandardEvaluationContext prepareEvaluationContextToUse(Object resource) {
StandardEvaluationContext evaluationContextToUse;
if (resource != null) {
evaluationContextToUse = this.createEvaluationContext();
if (resource instanceof MessageSourceResourceHolder) {
MessageSourceResourceHolder holder = (MessageSourceResourceHolder) 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 {
if (this.evaluationContext == null) {
this.evaluationContext = this.createEvaluationContext();
}
evaluationContextToUse = this.evaluationContext;
}
return evaluationContextToUse;
}
protected StandardEvaluationContext createEvaluationContext(){
if (this.getBeanFactory() != null) {
return ExpressionUtils.createStandardEvaluationContext(new BeanFactoryResolver(this.getBeanFactory()),
this.getConversionService());
}
else {
return ExpressionUtils.createStandardEvaluationContext(this.getConversionService());
}
}
}

86
spring-integration-core/src/main/java/org/springframework/integration/transaction/MessageSourceResourceHolder.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* 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.transaction;
import java.util.Collections;
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;
/**
* An implementation of the {@link ResourceHolder} which holds an instance of the current Message
* and the synchronization resource
*
* @author Gary Russell
* @author Oleg Zhurakousky
* @since 2.2
*
*/
public class MessageSourceResourceHolder implements ResourceHolder {
private final MessageSource<?> source;
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;
}
public Message<?> getMessage() {
return message;
}
/**
* Adds attribute to this {@link ResourceHolder} instance
*
* @param key
* @param value
*/
public void addAttribute(String key, Object value){
this.attributes.put(key, value);
}
/**
* Will return an immutable Mpa of current attributes.
* If you need to add attribute use {{@link #addAttribute(String, Object)} method.
*
* @return
*/
public Map<String, Object> getAttributes() {
return Collections.unmodifiableMap(attributes);
}
public void reset() {
}
public void unbound() {
}
public boolean isVoid() {
return false;
}
}

54
spring-integration-core/src/main/java/org/springframework/integration/transaction/PseudoTransactionManager.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* 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.transaction;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.support.AbstractPlatformTransactionManager;
import org.springframework.transaction.support.DefaultTransactionStatus;
/**
* An implementation of {@link PlatformTransactionManager} that provides transaction-like semantics to
* {@link MessageSource}s sources that are not inherently transactional. It does <b>not<b> make such
* sources transactional; rather, together with the <transaction-synchronization> element, it provides
* the ability to synchronize operations after a flow completes, via onSucess and onFailure expressions.
*
* @author Gary Russell
* @author Oleg Zhurakousky
* @since 2.2
*
*/
public class PseudoTransactionManager extends AbstractPlatformTransactionManager {
private static final long serialVersionUID = 1L;
@Override
protected Object doGetTransaction() throws TransactionException {
return new Object();
}
@Override
protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
//noop
}
@Override
protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
//noop
}
@Override
protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
//noop
}
}

26
spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactory.java Normal file
View File

@@ -0,0 +1,26 @@
/*
* 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.transaction;
import org.springframework.transaction.support.TransactionSynchronization;
/**
* Strategy for implementing factories that create {@link TransactionSynchronization}
*
* @author Oleg Zhurakousky
* @since 2.2
*
*/
public interface TransactionSynchronizationFactory {
TransactionSynchronization create(Object key);
}

31
spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* 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.transaction;
/**
* Strategy for implementing transaction synchronization processors
*
* @author Oleg Zhurakousky
* @author Gary Russell
* @since 2.2
*
*/
public interface TransactionSynchronizationProcessor {
public abstract void processBeforeCommit(MessageSourceResourceHolder holder);
public abstract void processAfterCommit(MessageSourceResourceHolder holder);
public abstract void processAfterRollback(MessageSourceResourceHolder holder);
}

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

@@ -34,6 +34,66 @@
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="transaction-synchronization-factory">
<xsd:annotation>
<xsd:documentation>
Allows you to configure org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory
This implementation of org.springframework.integration.transaction.TransactionSynchronizationFactory
allows you to configure SpEL expressions, with their execution being coordinated (synchronized) with a
transaction - see {TransactionSynchronization}. Expressions for before-commit, after.-commit, and after-rollback
are supported, together with a channel for each where the evaluation result (if any) will be sent.
For each sub-element you can specify 'expression' and/or 'channel' attributes.
If only the 'channel' attribute is present the received Message will be sent there as part of a particular synchronization scenario.
If only the 'expression' attribute is present and the result of an expression is a non-Null value, a Message with the
result as the payload will be generated and sent to a default channel (NullChannel) and will appear in the logs.
If you want the evaluation result to go to a specific channel add a 'channel' attribute. If the result of an expression is null
or void, no Message will be generated.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element name="before-commit" minOccurs="0" maxOccurs="1">
<xsd:complexType>
<xsd:attributeGroup ref="synchronizationAttributeGroup"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="after-commit" minOccurs="0" maxOccurs="1">
<xsd:complexType>
<xsd:attributeGroup ref="synchronizationAttributeGroup"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="after-rollback" minOccurs="0" maxOccurs="1">
<xsd:complexType>
<xsd:attributeGroup ref="synchronizationAttributeGroup"/>
</xsd:complexType>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
<xsd:attributeGroup name="synchronizationAttributeGroup">
<xsd:attribute name="expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
SpEL expression to be executed as part of Transaction synchronization
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation><![CDATA[
Reference to a MessageChannel where the result or an expression or received Message wil be sent.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:attributeGroup>
<xsd:element name="application-event-multicaster">
<xsd:complexType>
@@ -1494,34 +1554,6 @@
<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1" />
</xsd:choice>
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="pseudo-transactional" type="pseudoTransactionalType" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
This element is used to provide expressions that will be executed on success or failure,
when sending a message resulting from a poll. It provides semantics similar to a transaction
but users need to be aware that it is not a true transaction because the underlying resource
is not transactional.
The element is not allowed when a <transactional/> element is present.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="transaction-synchronization" type="pseudoTransactionalType" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
This element is used with <transactional/> to provide expressions that will be executed
on success or failure, when sending a message resulting from a poll.
The execution of the expressions is synchronized
with the encompassing transaction in that the onSuccess expression will be evaluated
immediately after the commit or the onFailure expression will be evaluated immediately
after the rollback. Users need to be aware that transaction synchronization does not
make an inherently non-transactional resource transactional, it simply implements the
best-effort one phase commit pattern.
The element is only allowed when a <transactional/> or <advice-chain/> element is present.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
</xsd:sequence>
<xsd:attribute name="fixed-delay" type="xsd:string">
<xsd:annotation>
@@ -3417,6 +3449,19 @@ is provided, the return value is expected to match a channel name exactly.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="synchronization-factory" type="xsd:string" >
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to an instance of org.springframework.integration.transaction.TransactionSynchronizationFactory
which will return an instance of org.springframework.transaction.support.TransactionSynchronization via its create(..) method.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.transaction.TransactionSynchronizationFactory" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="pseudoTransactionalType">
@@ -3442,6 +3487,20 @@ is provided, the return value is expected to match a channel name exactly.
<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded" />
</xsd:choice>
</xsd:sequence>
<xsd:attribute name="synchronization-factory" type="xsd:string" >
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to an instance of org.springframework.integration.transaction.TransactionSynchronizationFactory
which will return an instance of org.springframework.transaction.support.TransactionSynchronization via its create(..) method.
Setting of this attribute will only have an affect if a Transaction advice is present in the chain.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.transaction.TransactionSynchronizationFactory" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="expressionOrInnerEndpointDefinitionAware">

17
spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-config.xml Normal file
View File

@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.2.xsd">
<int:transaction-synchronization-factory id="syncFactoryComplete">
<int:before-commit channel="beforeCommitChannel"/>
<int:after-commit expression="'afterCommit'"/>
<int:after-rollback channel="afterRollbackChannel" expression="'afterRollback'"/>
</int:transaction-synchronization-factory>
<int:channel id="beforeCommitChannel"/>
<int:channel id="afterRollbackChannel"/>
</beans>

41
spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-xsd.xml Normal file
View File

@@ -0,0 +1,41 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.2.xsd">
<int:transaction-synchronization-factory id="a">
<int:before-commit channel="nullChannel"/>
</int:transaction-synchronization-factory>
<int:transaction-synchronization-factory id="b">
<int:after-commit expression="'foo'"/>
</int:transaction-synchronization-factory>
<int:transaction-synchronization-factory id="c">
<int:after-rollback expression="'f'"/>
</int:transaction-synchronization-factory>
<int:transaction-synchronization-factory id="d">
<int:before-commit channel="nullChannel" expression="''"/>
<int:after-commit channel="nullChannel" expression="''"/>
</int:transaction-synchronization-factory>
<int:transaction-synchronization-factory id="e">
<int:before-commit channel="nullChannel" expression="''"/>
<int:after-rollback channel="nullChannel" expression="''"/>
</int:transaction-synchronization-factory>
<int:transaction-synchronization-factory id="f">
<int:after-commit expression="'f'"/>
<int:after-rollback channel="nullChannel"/>
</int:transaction-synchronization-factory>
<int:transaction-synchronization-factory id="g">
<int:before-commit channel="nullChannel"/>
<int:after-commit expression="'f'"/>
<int:after-rollback channel="nullChannel" expression="''"/>
</int:transaction-synchronization-factory>
</beans>

72
spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests.java Normal file
View File

@@ -0,0 +1,72 @@
/*
* 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.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory;
import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor;
import org.springframework.integration.transaction.TransactionSynchronizationProcessor;
/**
* @author Oleg Zhurakousky
*/
public class TransactionSynchronizationFactoryParserTests {
@Test // nothing to assert. Validates only XSD
public void validateXsdCombinationOfOrderOfSubelements(){
new ClassPathXmlApplicationContext("TransactionSynchronizationFactoryParserTests-xsd.xml", this.getClass());
}
@Test
public void validateFullConfiguration(){
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("TransactionSynchronizationFactoryParserTests-config.xml", this.getClass());
DefaultTransactionSynchronizationFactory syncFactory =
context.getBean("syncFactoryComplete", DefaultTransactionSynchronizationFactory.class);
assertNotNull(syncFactory);
TransactionSynchronizationProcessor processor =
TestUtils.getPropertyValue(syncFactory, "processor", ExpressionEvaluatingTransactionSynchronizationProcessor.class);
assertNotNull(processor);
MessageChannel beforeCommitResultChannel = TestUtils.getPropertyValue(processor, "beforeCommitChannel", MessageChannel.class);
assertNotNull(beforeCommitResultChannel);
assertEquals(beforeCommitResultChannel, context.getBean("beforeCommitChannel"));
Object beforeCommitExpression = TestUtils.getPropertyValue(processor, "beforeCommitExpression");
assertNull(beforeCommitExpression);
MessageChannel afterCommitResultChannel = TestUtils.getPropertyValue(processor, "afterCommitChannel", MessageChannel.class);
assertNotNull(afterCommitResultChannel);
assertEquals(afterCommitResultChannel, context.getBean("nullChannel"));
Expression afterCommitExpression = TestUtils.getPropertyValue(processor, "afterCommitExpression", Expression.class);
assertNotNull(afterCommitExpression);
assertEquals("'afterCommit'", ((SpelExpression)afterCommitExpression).getExpressionString());
MessageChannel afterRollbackResultChannel = TestUtils.getPropertyValue(processor, "afterRollbackChannel", MessageChannel.class);
assertNotNull(afterRollbackResultChannel);
assertEquals(afterRollbackResultChannel, context.getBean("afterRollbackChannel"));
Expression afterRollbackExpression = TestUtils.getPropertyValue(processor, "afterRollbackExpression", Expression.class);
assertNotNull(afterRollbackExpression);
assertEquals("'afterRollback'", ((SpelExpression)afterRollbackExpression).getExpressionString());
}
}

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

@@ -17,24 +17,28 @@ package org.springframework.integration.endpoint;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
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.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PseudoTransactionalMessageSource;
import org.springframework.integration.core.MessageSource;
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.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionSynchronizationUtils;
import org.springframework.transaction.support.TransactionTemplate;
/**
* @author Gary Russell
* @author Oleg Zhurakousky
* @since 2.2
*
*/
@@ -43,155 +47,69 @@ public class PseudoTransactionalMessageSourceTests {
@Test
public void testCommit() {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
PollableChannel queueChannel = new QueueChannel();
syncProcessor.setBeforeCommitExpression(new SpelExpressionParser().parseExpression("#bix"));
syncProcessor.setBeforeCommitChannel(queueChannel);
syncProcessor.setAfterCommitChannel(queueChannel);
syncProcessor.setAfterCommitExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
final Object object = new Object();
final AtomicReference<Object> committed = new AtomicReference<Object>();
final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
adapter.setSource(new MessageSource<String>() {
public Message<String> receive() {
return new GenericMessage<String>("foo");
}
public Object getResource() {
return object;
}
public void afterCommit(Object resource) {
committed.set(resource);
}
public void afterRollback(Object resource) {
rolledBack.set(resource);
}
public void afterReceiveNoTx(Object resource) {
}
public void afterSendNoTx(Object resource) {
GenericMessage<String> message = new GenericMessage<String>("foo");
((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux");
((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("bix", "qox");
return message;
}
});
TransactionSynchronizationManager.initSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(true);
adapter.doPoll();
TransactionSynchronizationUtils.triggerBeforeCommit(false);
TransactionSynchronizationUtils.triggerAfterCommit();
assertSame(object, committed.get());
Message<?> beforeCommitMessage = queueChannel.receive(1000);
assertNotNull(beforeCommitMessage);
assertEquals("qox", beforeCommitMessage.getPayload());
Message<?> afterCommitMessage = queueChannel.receive(1000);
assertNotNull(afterCommitMessage);
assertEquals("qux", afterCommitMessage.getPayload());
TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
TransactionSynchronizationManager.clearSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(false);
assertNull(rolledBack.get());
}
@Test
public void testPseudoCommitWithMessage() {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
final Object object = new Object();
final AtomicReference<Object> afterReceive = new AtomicReference<Object>();
final AtomicReference<Object> afterSend = new AtomicReference<Object>();
adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
public Message<String> receive() {
return new GenericMessage<String>("foo");
}
public Object getResource() {
return object;
}
public void afterCommit(Object resource) {
throw new RuntimeException("no tx - commit not expected");
}
public void afterRollback(Object resource) {
throw new RuntimeException("no tx - rollback not expected");
}
public void afterReceiveNoTx(Object resource) {
afterReceive.set(resource);
}
public void afterSendNoTx(Object resource) {
afterSend.set(resource);
}
});
adapter.doPoll();
assertSame(object, afterReceive.get());
assertSame(object, afterSend.get());
}
@Test
public void testPseudoCommitNoMessage() {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
final Object object = new Object();
final AtomicReference<Object> afterReceive = new AtomicReference<Object>();
adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
public Message<String> receive() {
return null;
}
public Object getResource() {
return object;
}
public void afterCommit(Object resource) {
throw new RuntimeException("no tx - commit not expected");
}
public void afterRollback(Object resource) {
throw new RuntimeException("no tx - rollback not expected");
}
public void afterReceiveNoTx(Object resource) {
afterReceive.set(resource);
}
public void afterSendNoTx(Object resource) {
throw new RuntimeException("no message - after send not expected");
}
});
adapter.doPoll();
assertSame(object, afterReceive.get());
}
@Test
public void testRollback() {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
PollableChannel queueChannel = new QueueChannel();
syncProcessor.setAfterRollbackChannel(queueChannel);
syncProcessor.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
final Object object = new Object();
final AtomicReference<Object> committed = new AtomicReference<Object>();
final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
adapter.setSource(new MessageSource<String>() {
public Message<String> receive() {
return new GenericMessage<String>("foo");
}
public Object getResource() {
return object;
}
public void afterCommit(Object resource) {
committed.set(resource);
}
public void afterRollback(Object resource) {
rolledBack.set(resource);
}
public void afterReceiveNoTx(Object resource) {
}
public void afterSendNoTx(Object resource) {
GenericMessage<String> message = new GenericMessage<String>("foo");
((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux");
return message;
}
});
@@ -199,77 +117,139 @@ public class PseudoTransactionalMessageSourceTests {
TransactionSynchronizationManager.setActualTransactionActive(true);
adapter.doPoll();
TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
assertSame(object, rolledBack.get());
Message<?> rollbackMessage = queueChannel.receive(1000);
assertNotNull(rollbackMessage);
assertEquals("qux", rollbackMessage.getPayload());
TransactionSynchronizationManager.clearSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(false);
assertNull(committed.get());
}
@Test
public void testSuccessAndFailureEvaluationWithResource() {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
final Object object = new Bar();
final AtomicReference<Object> committed = new AtomicReference<Object>();
final AtomicReference<Object> rolledBack = new AtomicReference<Object>();
adapter.setSource(new PseudoTransactionalMessageSource<String, Object>() {
public void testCommitWithManager() {
final PollableChannel queueChannel = new QueueChannel();
TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager());
transactionTemplate.execute(new TransactionCallback<Object>() {
public Message<String> receive() {
return new GenericMessage<String>("foo");
}
public Object doInTransaction(TransactionStatus status) {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setBeforeCommitExpression(new SpelExpressionParser().parseExpression("#bix"));
syncProcessor.setBeforeCommitChannel(queueChannel);
syncProcessor.setAfterCommitChannel(queueChannel);
syncProcessor.setAfterCommitExpression(new SpelExpressionParser().parseExpression("#baz"));
public Object getResource() {
return object;
}
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
public void afterCommit(Object resource) {
committed.set(resource);
}
adapter.setTransactionSynchronizationFactory(syncFactory);
public void afterRollback(Object resource) {
rolledBack.set(resource);
}
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setSource(new MessageSource<String>() {
public void afterReceiveNoTx(Object resource) {
}
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");
return message;
}
});
public void afterSendNoTx(Object resource) {
adapter.doPoll();
return null;
}
});
Message<?> beforeCommitMessage = queueChannel.receive(1000);
assertNotNull(beforeCommitMessage);
assertEquals("qox", beforeCommitMessage.getPayload());
Message<?> afterCommitMessage = queueChannel.receive(1000);
assertNotNull(afterCommitMessage);
assertEquals("qux", afterCommitMessage.getPayload());
}
TransactionSynchronizationManager.initSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(true);
adapter.setOnSuccessExpression(new SpelExpressionParser().parseExpression("payload + #resource.value"));
QueueChannel success = new QueueChannel();
adapter.setOnSuccessResultChannel(success);
adapter.setOnFailureExpression(new SpelExpressionParser().parseExpression("payload + 'X' + #resource.value"));
QueueChannel failure = new QueueChannel();
adapter.setOnFailureChannel(failure);
@Test
public void testRollbackWithManager() {
final PollableChannel queueChannel = new QueueChannel();
TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager());
try {
transactionTemplate.execute(new TransactionCallback<Object>() {
adapter.doPoll();
TransactionSynchronizationUtils.triggerAfterCommit();
assertSame(object, committed.get());
TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
TransactionSynchronizationManager.clearSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(false);
assertNull(rolledBack.get());
Message<?> result = success.receive(10000);
assertNotNull(result);
assertEquals("foobar", result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
committed.set(null);
public Object doInTransaction(TransactionStatus status) {
TransactionSynchronizationManager.initSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(true);
adapter.doPoll();
TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
assertSame(object, rolledBack.get());
TransactionSynchronizationManager.clearSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(false);
assertNull(committed.get());
result = failure.receive(10000);
assertNotNull(result);
assertEquals("fooXbar", result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor = new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setAfterRollbackChannel(queueChannel);
syncProcessor.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory = new DefaultTransactionSynchronizationFactory(
syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setSource(new MessageSource<String>() {
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
return message;
}
});
adapter.doPoll();
throw new RuntimeException("Force rollback");
}
});
}
catch (Exception e) {
assertEquals("Force rollback", e.getMessage());
}
Message<?> rollbackMessage = queueChannel.receive(1000);
assertNotNull(rollbackMessage);
assertEquals("qux", rollbackMessage.getPayload());
}
@Test
public void testRollbackWithManagerUsingStatus() {
final PollableChannel queueChannel = new QueueChannel();
TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager());
transactionTemplate.execute(new TransactionCallback<Object>() {
public Object doInTransaction(TransactionStatus status) {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor = new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setAfterRollbackChannel(queueChannel);
syncProcessor.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory = new DefaultTransactionSynchronizationFactory(
syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setSource(new MessageSource<String>() {
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((MessageSourceResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
return message;
}
});
adapter.doPoll();
status.setRollbackOnly();
return null;
}
});
Message<?> rollbackMessage = queueChannel.receive(1000);
assertNotNull(rollbackMessage);
assertEquals("qux", rollbackMessage.getPayload());
}
public class Bar {

27
spring-integration-file/.springBeans
View File

@@ -1,13 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.5.0.201008251000-M3]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<configs>
</configs>
<configSets>
</configSets>
</beansProjectDescription>
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.9.2.201205070117-RELEASE]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<configs>
<config>src/test/java/org/springframework/integration/file/config/config.xml</config>
</configs>
<configSets>
</configSets>
</beansProjectDescription>

36
spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml
View File

@@ -14,10 +14,7 @@
<int-file:inbound-channel-adapter id="pseudoTx"
channel="input" auto-startup="false" directory="${java.io.tmpdir}/si-test1">
<int:poller fixed-rate="500">
<int:pseudo-transactional
on-success-expression="payload.delete()" on-success-result-channel="successChannel"
on-failure-expression="'foo'" on-failure-result-channel="failureChannel"
send-timeout="500" />
<int:transactional synchronization-factory="syncFactoryA"/>
</int:poller>
</int-file:inbound-channel-adapter>
@@ -36,15 +33,34 @@
<int-file:inbound-channel-adapter id="realTx" channel="txInput" auto-startup="false"
directory="${java.io.tmpdir}/si-test2">
<int:poller fixed-rate="500">
<int:transactional transaction-manager="txManager" />
<int:transaction-synchronization on-success-expression="@txManager.committed"
on-success-result-channel="successChannel"
on-failure-expression="@txManager.rolledBack"
on-failure-result-channel="failureChannel"
send-timeout="5000" />
<int:transactional transaction-manager="txManager" synchronization-factory="syncFactoryB"/>
</int:poller>
</int-file:inbound-channel-adapter>
<bean id="txManager" class="org.springframework.integration.file.FileInboundTransactionTests$DummyTxManager" />
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
<bean id="syncFactoryA" class="org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory">
<constructor-arg>
<bean class="org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor">
<property name="afterCommitExpression" value="#{new org.springframework.expression.spel.standard.SpelExpressionParser().parseExpression('payload.delete()')}"/>
<property name="afterCommitChannel" ref="successChannel"/>
<property name="afterRollbackExpression" value="#{new org.springframework.expression.common.LiteralExpression('foo')}"/>
<property name="afterRollbackChannel" ref="failureChannel"/>
</bean>
</constructor-arg>
</bean>
<bean id="syncFactoryB" class="org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory">
<constructor-arg>
<bean class="org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor">
<property name="afterCommitExpression" value="#{new org.springframework.expression.spel.standard.SpelExpressionParser().parseExpression('@txManager.committed')}"/>
<property name="afterCommitChannel" ref="successChannel"/>
<property name="afterRollbackExpression" value="#{new org.springframework.expression.spel.standard.SpelExpressionParser().parseExpression('@txManager.rolledBack')}"/>
<property name="afterRollbackChannel" ref="failureChannel"/>
</bean>
</constructor-arg>
</bean>
</beans>

10
spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java
View File

@@ -26,10 +26,10 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
@@ -95,7 +95,7 @@ public class FileInboundTransactionTests {
file.createNewFile();
Message<?> result = successChannel.receive(10000);
assertNotNull(result);
assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
assertEquals(Boolean.TRUE, result.getPayload());
System.out.println(result);
assertFalse(file.delete());
crash.set(true);
@@ -105,7 +105,7 @@ public class FileInboundTransactionTests {
assertNotNull(result);
System.out.println(result);
assertTrue(file.delete());
assertEquals("foo", result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
assertEquals("foo", result.getPayload());
pseudoTx.stop();
assertFalse(transactionManager.getCommitted());
assertFalse(transactionManager.getRolledBack());
@@ -131,7 +131,7 @@ public class FileInboundTransactionTests {
file.createNewFile();
Message<?> result = successChannel.receive(10000);
assertNotNull(result);
assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
assertEquals(Boolean.TRUE, result.getPayload());
assertTrue(file.delete());
System.out.println(result);
assertTrue(transactionManager.getCommitted());
@@ -142,7 +142,7 @@ public class FileInboundTransactionTests {
assertNotNull(result);
System.out.println(result);
assertTrue(file.delete());
assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
assertEquals(Boolean.TRUE, result.getPayload());
realTx.stop();
assertTrue(transactionManager.getRolledBack());
}

14
spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml
View File

@@ -36,8 +36,7 @@
</bean>
<si:poller default="true" fixed-rate="10">
<si:pseudo-transactional on-success-expression="payload.delete()"
on-success-result-channel="resultChannel" />
<si:transactional synchronization-factory="syncFactory"/>
</si:poller>
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
@@ -45,4 +44,15 @@
<si:channel id="resultChannel">
<si:queue />
</si:channel>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
<bean id="syncFactory" class="org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory">
<constructor-arg>
<bean class="org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor">
<property name="afterCommitExpression" value="#{new org.springframework.expression.spel.standard.SpelExpressionParser().parseExpression('payload.delete()')}"/>
<property name="afterCommitChannel" ref="resultChannel"/>
</bean>
</constructor-arg>
</bean>
</beans>

6
spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java
View File

@@ -25,9 +25,9 @@ import java.io.File;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.core.PollableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -47,7 +47,7 @@ public class FileToChannelIntegrationTests {
@Autowired
PollableChannel resultChannel;
@Test(timeout = 2000)
@Test//(timeout = 2000)
public void fileMessageToChannel() throws Exception {
File file = File.createTempFile("test", null, inputDirectory);
file.setLastModified(System.currentTimeMillis() - 1000);
@@ -59,7 +59,7 @@ public class FileToChannelIntegrationTests {
assertNotNull(received.getPayload());
Message<?> result = resultChannel.receive(10000);
assertNotNull(result);
assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT));
assertEquals(Boolean.TRUE, result.getPayload());
assertTrue(!file.exists());
}

19
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml
View File

@@ -16,11 +16,7 @@
comparator="testComparator"
auto-startup="false">
<integration:poller fixed-rate="5000">
<integration:pseudo-transactional on-success-expression="payload.delete()"
on-success-result-channel="successChannel"
on-failure-expression="'foo'"
on-failure-result-channel="nullChannel"
send-timeout="5000" />
<integration:transactional synchronization-factory="syncFactory"/>
</integration:poller>
</inbound-channel-adapter>
@@ -38,5 +34,18 @@
<beans:bean id="testComparator"
class="org.springframework.integration.file.config.FileInboundChannelAdapterParserTests$TestComparator"/>
<beans:bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
<beans:bean id="syncFactory" class="org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory">
<beans:constructor-arg>
<beans:bean class="org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor">
<beans:property name="afterCommitExpression" value="#{new org.springframework.expression.spel.standard.SpelExpressionParser().parseExpression('payload.delete()')}"/>
<beans:property name="afterCommitChannel" ref="successChannel"/>
<beans:property name="afterRollbackExpression" value="#{new org.springframework.expression.common.LiteralExpression('foo')}"/>
<beans:property name="afterRollbackChannel" ref="nullChannel"/>
</beans:bean>
</beans:constructor-arg>
</beans:bean>
</beans:beans>

1
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java
View File

@@ -27,6 +27,7 @@ import java.util.concurrent.PriorityBlockingQueue;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

13
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml
View File

@@ -26,13 +26,14 @@
temporary-file-suffix=".foo"
remote-directory="foo/bar">
<int:poller fixed-rate="1000">
<int:pseudo-transactional on-success-expression="'foo'"
on-success-result-channel="successChannel"
on-failure-expression="'bar'"
on-failure-result-channel="failureChannel"
send-timeout="123" />
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-ftp:inbound-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="'foo'" channel="successChannel"/>
<int:after-rollback expression="'bar'" channel="failureChannel"/>
</int:transaction-synchronization-factory>
<int:channel id="successChannel" />
@@ -78,5 +79,7 @@
</int-ftp:inbound-channel-adapter>
<int:bridge input-channel="autoChannel" output-channel="nullChannel" />
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
</beans>

12
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
View File

@@ -29,10 +29,10 @@ import java.util.Comparator;
import java.util.Map;
import org.junit.Test;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
@@ -77,16 +77,6 @@ public class FtpInboundChannelAdapterParserTests {
assertNotNull(filter);
Object sessionFactory = TestUtils.getPropertyValue(fisync, "sessionFactory");
assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass()));
assertEquals("foo", TestUtils.getPropertyValue(adapter, "onSuccessExpression", Expression.class).getValue());
assertSame(ac.getBean("successChannel"), TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "defaultChannel"));
assertEquals(123L, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "sendTimeout"));
assertEquals("bar", TestUtils.getPropertyValue(adapter, "onFailureExpression", Expression.class).getValue());
assertSame(ac.getBean("failureChannel"), TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "defaultChannel"));
assertEquals(123L, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "sendTimeout"));
}
@Test

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

@@ -1,27 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
<int:inbound-channel-adapter channel="input">
<bean class="org.springframework.integration.jdbc.PseudoTransactionalMessageSourceTests$MessageSource"/>
<int:poller fixed-delay="2000">
<int:transactional />
<int:transaction-synchronization/>
</int:poller>
</int:inbound-channel-adapter>
<int:channel id="input" />
<jdbc:embedded-database id="dataSource" type="HSQL" />
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource" />
</bean>
</beans>

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

@@ -1,109 +0,0 @@
/*
* 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.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PseudoTransactionalMessageSource;
import org.springframework.integration.core.SubscribableChannel;
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;
@Autowired
private SubscribableChannel input;
@Test
public void testCommit() throws Exception {
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
}
};
input.subscribe(handler);
assertTrue(latch1.await(10, TimeUnit.SECONDS));
assertTrue(committed);
input.unsubscribe(handler);
}
@Test
public void testRollback() throws Exception {
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("expected");
}
};
input.subscribe(handler);
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertTrue(rolledBack);
input.unsubscribe(handler);
}
public static class MessageSource implements PseudoTransactionalMessageSource<String, Object> {
public Message<String> receive() {
return new GenericMessage<String>("foo");
}
public Object getResource() {
return new Object();
}
public void afterCommit(Object resource) {
committed = true;
latch1.countDown();
}
public void afterRollback(Object resource) {
rolledBack = true;
latch2.countDown();
}
public void afterReceiveNoTx(Object resource) {
}
public void afterSendNoTx(Object resource) {
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2011 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,6 +33,7 @@ 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;
@@ -41,18 +42,17 @@ 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 final ThreadLocal<MailReceiverContext> contextHolder = new ThreadLocal<MailReceiverContext>();
private volatile Folder folder;
private volatile boolean shouldDeleteMessages;
protected volatile int folderOpenMode = Folder.READ_ONLY;
private volatile Properties javaMailProperties = new Properties();
@@ -81,6 +81,8 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
protected volatile boolean initialized;
private final Object folderMonitor = new Object();
public AbstractMailReceiver() {
this.url = null;
@@ -115,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)
*/
@@ -127,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)
*/
@@ -138,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) {
@@ -166,27 +168,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
}
protected Folder getFolder() {
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();
return this.contextHolder.get();
}
catch (MessagingException e) {
throw new org.springframework.integration.MessagingException("Failed to open folder", e);
}
}
return mailReceiverContext;
return this.folder;
}
/**
@@ -222,71 +204,104 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
}
}
protected synchronized void openFolder() throws MessagingException {
protected void openFolder() throws MessagingException {
this.openSession();
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 = this.store.getFolder(this.url);
}
else {
folder = context.getFolder();
}
if (folder == null || !folder.exists()) {
if (this.folder == null || !this.folder.exists()) {
throw new IllegalStateException("no such folder [" + this.url.getFile() + "]");
}
if (folder.isOpen()) {
if (this.folder.isOpen()) {
return;
}
if (logger.isDebugEnabled()) {
logger.debug("opening folder [" + MailTransportUtils.toPasswordProtectedString(this.url) + "]");
}
folder.open(this.folderOpenMode);
this.folder.open(this.folderOpenMode);
}
public Message[] receive() throws javax.mail.MessagingException {
this.openFolder();
if (logger.isInfoEnabled()) {
logger.info("attempting to receive mail from folder [" + this.getFolder().getFullName() + "]");
}
Message[] messages = this.searchForNewMessages();
if (this.maxFetchSize > 0 && messages.length > this.maxFetchSize) {
Message[] reducedMessages = new Message[this.maxFetchSize];
System.arraycopy(messages, 0, reducedMessages, 0, this.maxFetchSize);
messages = reducedMessages;
}
if (logger.isDebugEnabled()) {
logger.debug("found " + messages.length + " new messages");
}
if (messages.length > 0) {
this.fetchMessages(messages);
}
List<Message> copiedMessages = new LinkedList<Message>();
logger.debug("Received " + 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));
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() + "]");
}
Message[] messages = this.searchForNewMessages();
if (this.maxFetchSize > 0 && messages.length > this.maxFetchSize) {
Message[] reducedMessages = new Message[this.maxFetchSize];
System.arraycopy(messages, 0, reducedMessages, 0, this.maxFetchSize);
messages = reducedMessages;
}
if (logger.isDebugEnabled()) {
logger.debug("found " + messages.length + " new messages");
}
if (messages.length > 0) {
this.fetchMessages(messages);
}
List<Message> copiedMessages = new LinkedList<Message>();
logger.debug("Recieved " + messages.length + " messages");
boolean recentFlagSupported = false;
Flags flags = this.getFolder().getPermanentFlags();
if (flags != null){
recentFlagSupported = flags.contains(Flags.Flag.RECENT);
}
for (int i = 0; i < messages.length; i++) {
if (!recentFlagSupported){
if (flags != null && flags.contains(Flags.Flag.USER)){
if (logger.isDebugEnabled()){
logger.debug("USER flags are supported by this mail server. Flagging message with '" + SI_USER_FLAG + "' user flag");
}
Flags siFlags = new Flags();
siFlags.add(SI_USER_FLAG);
messages[i].setFlags(siFlags, true);
}
else {
if (logger.isDebugEnabled()){
logger.debug("USER flags are not supported by this mail server. Flagging message with system flag");
}
messages[i].setFlag(Flags.Flag.FLAGGED, true);
}
}
if (this.selectorExpression != null) {
Message message = messages[i];
if (this.selectorExpression.getValue(this.context, message, Boolean.class)){
this.setAdditionalFlags(message);
copiedMessages.add(new MimeMessage((MimeMessage) message));
}
else {
if (logger.isDebugEnabled()){
logger.debug("Fetched email with subject '" + message.getSubject() + "' will be discarded by the matching filter" +
" and will not be flagged as SEEN.");
}
}
}
else {
this.setAdditionalFlags(messages[i]);
copiedMessages.add(new MimeMessage((MimeMessage) messages[i]));
}
}
if (this.shouldDeleteMessages()) {
this.deleteMessages(messages);
}
return copiedMessages.toArray(new Message[copiedMessages.size()]);
}
else {
copiedMessages.add(new MimeMessage((MimeMessage) messages[i]));
finally {
MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
}
}
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
*/
@@ -295,12 +310,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.contextHolder.get().getFolder().fetch(messages, contentsProfile);
this.folder.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
*/
@@ -311,19 +326,23 @@ 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 synchronized void destroy() throws Exception {
MailTransportUtils.closeService(this.store);
this.store = null;
this.initialized = false;
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;
}
}
@Override
@@ -342,84 +361,4 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
return this.store;
}
/**
* Delete and expunge messages after success.
* @param context
*/
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.remove();
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.remove();
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2011 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,7 +19,6 @@ 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;
@@ -27,7 +26,6 @@ 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;
@@ -40,11 +38,10 @@ 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 {
@@ -61,7 +58,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
private volatile ScheduledFuture<?> pingTask;
private volatile long connectionPingInterval = 10000;
private final ExceptionAwarePeriodicTrigger receivingTaskTrigger = new ExceptionAwarePeriodicTrigger();
@@ -80,7 +77,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
this.shouldReconnectAutomatically = shouldReconnectAutomatically;
}
@Override
public String getComponentType() {
return "mail:imap-idle-channel-adapter";
}
@@ -134,16 +130,12 @@ 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();
context = mailReceiver.getTransactionContext();
Assert.state(context != null, "Mail receiver returned a null context");
Folder folder = context.getFolder();
if (folder.isOpen()) {
if (mailReceiver.getFolder().isOpen()) {
Message[] mailMessages = mailReceiver.receive();
if (logger.isDebugEnabled()) {
logger.debug("received " + mailMessages.length + " mail messages");
@@ -167,11 +159,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
"Failure in 'idle' task. Will NOT resubmit.", e);
}
}
finally {
if (context != null) {
mailReceiver.closeContextAfterSuccess(context);
}
}
}
}
@@ -189,9 +176,9 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
}
}
}
private class ExceptionAwarePeriodicTrigger implements Trigger {
private volatile boolean delayNextExecution;
@@ -199,12 +186,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;
}

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

@@ -32,12 +32,6 @@ 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;
@@ -60,6 +54,5 @@ public interface MailReceiver {
Folder getFolder() {
return folder;
}
}
}

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

@@ -25,8 +25,6 @@ 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;
@@ -38,8 +36,9 @@ import org.springframework.util.Assert;
* @author Jonas Partner
* @author Mark Fisher
* @author Gary Russell
* @author Oleg Zhurakousky
*/
public class MailReceivingMessageSource implements PseudoTransactionalMessageSource<javax.mail.Message, MailReceiverContext> {
public class MailReceivingMessageSource implements MessageSource<javax.mail.Message> {
private final Log logger = LogFactory.getLog(this.getClass());
@@ -53,7 +52,6 @@ public class MailReceivingMessageSource implements PseudoTransactionalMessageSou
this.mailReceiver = mailReceiver;
}
public Message<javax.mail.Message> receive() {
try {
javax.mail.Message mailMessage = this.mailQueue.poll();
@@ -77,31 +75,4 @@ public class MailReceivingMessageSource implements PseudoTransactionalMessageSou
return null;
}
public MailReceiverContext getResource() {
return this.mailReceiver.getTransactionContext();
}
public void afterCommit(Object context) {
Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
this.mailReceiver.closeContextAfterSuccess((MailReceiverContext) context);
}
public void afterRollback(Object context) {
Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext");
this.mailReceiver.closeContextAfterFailure((MailReceiverContext) context);
}
/**
* For backwards-compatibility; with no tx, the mail adapter updates the status before the send.
*/
public void afterReceiveNoTx(MailReceiverContext resource) {
this.afterCommit(resource);
}
/**
* For backwards-compatibility; with no tx, the mail adapter updates the status before the send.
*/
public void afterSendNoTx(MailReceiverContext resource) {
// No op
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2010 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,7 +17,6 @@ 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;
@@ -25,9 +24,8 @@ 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;
@@ -44,6 +42,7 @@ 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;
@@ -52,7 +51,6 @@ 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;
@@ -64,23 +62,25 @@ import com.sun.mail.imap.IMAPFolder;
*
*/
public class ImapMailReceiverTests {
private final AtomicInteger failed = new AtomicInteger(0);
private AtomicInteger failed = new AtomicInteger(0);
@Test
public void receiveAndMarkAsReadDontDelete() throws Exception{
AbstractMailReceiver receiver = new ImapMailReceiver();
((ImapMailReceiver)receiver).setShouldMarkMessagesAsRead(true);
receiver = spy(receiver);
receiver.afterPropertiesSet();
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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};
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
@@ -88,24 +88,23 @@ public class ImapMailReceiverTests {
if (folderOpenMode != Folder.READ_WRITE){
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
}
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).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 +116,13 @@ public class ImapMailReceiverTests {
receiver.setShouldDeleteMessages(true);
receiver = spy(receiver);
receiver.afterPropertiesSet();
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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};
@@ -135,20 +136,19 @@ public class ImapMailReceiverTests {
return null;
}
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).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());
@@ -159,12 +159,14 @@ public class ImapMailReceiverTests {
((ImapMailReceiver)receiver).setShouldMarkMessagesAsRead(false);
receiver = spy(receiver);
receiver.afterPropertiesSet();
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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};
@@ -173,13 +175,13 @@ public class ImapMailReceiverTests {
return null;
}
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
@@ -187,7 +189,6 @@ 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);
}
@@ -198,11 +199,13 @@ public class ImapMailReceiverTests {
((ImapMailReceiver)receiver).setShouldMarkMessagesAsRead(false);
receiver = spy(receiver);
receiver.afterPropertiesSet();
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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};
@@ -216,13 +219,13 @@ public class ImapMailReceiverTests {
return null;
}
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
@@ -230,7 +233,6 @@ 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);
@@ -241,11 +243,13 @@ public class ImapMailReceiverTests {
AbstractMailReceiver receiver = new ImapMailReceiver();
receiver = spy(receiver);
receiver.afterPropertiesSet();
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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};
@@ -259,20 +263,19 @@ public class ImapMailReceiverTests {
return null;
}
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).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());
@@ -280,22 +283,22 @@ public class ImapMailReceiverTests {
@Test
@Ignore
public void testMessageHistory() throws Exception{
ApplicationContext context =
ApplicationContext context =
new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class);
ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class);
AbstractMailReceiver receiver = new ImapMailReceiver();
receiver = spy(receiver);
receiver.afterPropertiesSet();
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
adapterAccessor.setPropertyValue("mailReceiver", receiver);
MimeMessage mailMessage = mock(MimeMessage.class);
Flags flags = mock(Flags.class);
when(mailMessage.getFlags()).thenReturn(flags);
final Message[] messages = new Message[]{mailMessage};
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
DirectFieldAccessor accesor = new DirectFieldAccessor((invocation.getMock()));
@@ -305,19 +308,19 @@ public class ImapMailReceiverTests {
return null;
}
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
PollableChannel channel = context.getBean("channel", PollableChannel.class);
adapter.start();
@@ -331,17 +334,16 @@ public class ImapMailReceiverTests {
@Test
public void testIdleChannelAdapterException() throws Exception{
ApplicationContext context =
ApplicationContext context =
new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class);
ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class);
//ImapMailReceiver receiver = (ImapMailReceiver) TestUtils.getPropertyValue(adapter, "mailReceiver");
DirectChannel channel = new DirectChannel();
channel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(org.springframework.integration.Message<?> requestMessage) {
throw new RuntimeException("Failed");
}
@@ -349,64 +351,59 @@ public class ImapMailReceiverTests {
adapter.setOutputChannel(channel);
QueueChannel errorChannel = new QueueChannel();
adapter.setErrorChannel(errorChannel);
AbstractMailReceiver receiver = new ImapMailReceiver();
receiver = spy(receiver);
receiver.afterPropertiesSet();
@SuppressWarnings("unchecked")
final ThreadLocal<MailReceiverContext> contextHolder = TestUtils.getPropertyValue(receiver, "contextHolder", ThreadLocal.class);
final Folder folder = mock(IMAPFolder.class);
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(IMAPFolder.class);
when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
folderField.set(receiver, folder);
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return true;
}
}).when(folder).isOpen();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
contextHolder.set(new MailReceiverContext(folder));
return null;
}
}).when(receiver).openFolder();
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
adapterAccessor.setPropertyValue("mailReceiver", receiver);
MimeMessage mailMessage = mock(MimeMessage.class);
Flags flags = mock(Flags.class);
when(mailMessage.getFlags()).thenReturn(flags);
final Message[] messages = new Message[]{mailMessage};
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
adapter.start();
org.springframework.integration.Message<?> replMessage = errorChannel.receive(10000);
assertNotNull(replMessage);
assertEquals("Failed", ((Exception) replMessage.getPayload()).getCause().getMessage());
}
@Test // see INT-1801
public void testImapLifecycleForRaceCondition() throws Exception{
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++) {
for (int i = 0; i < 1000; i++) {
final ImapMailReceiver receiver = new ImapMailReceiver("imap://foo");
Store store = mock(Store.class);
Folder folder = mock(Folder.class);
@@ -415,12 +412,12 @@ public class ImapMailReceiverTests {
when(folder.search((SearchTerm) Mockito.any())).thenReturn(new Message[]{});
when(store.getFolder(Mockito.any(URLName.class))).thenReturn(folder);
when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
DirectFieldAccessor df = new DirectFieldAccessor(receiver);
df.setPropertyValue("store", store);
receiver.afterPropertiesSet();
new Thread(new Runnable() {
public void run(){
try {
@@ -431,11 +428,10 @@ public class ImapMailReceiverTests {
failed.getAndIncrement();
}
}
receiveCount.incrementAndGet();
receiveLatch.countDown();
}
}).start();
new Thread(new Runnable() {
public void run(){
try {
@@ -444,23 +440,9 @@ 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());
}
@Test
public void testCustomSearchTermStrategy() throws Exception{
ImapMailReceiver receiver = new ImapMailReceiver();
SearchTermStrategy stStrategy = mock(SearchTermStrategy.class);
receiver.setSearchTermStrategy(stStrategy);
assertEquals(stStrategy, TestUtils.getPropertyValue(receiver, "searchTermStrategy"));
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2010 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,8 +17,10 @@ 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;
@@ -30,12 +32,11 @@ 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 {
@@ -44,12 +45,13 @@ public class ImapMailSearchTermsTests {
public void validateSearchTermsWhenShouldMarkAsReadNoExistingFlags() throws Exception {
ImapMailReceiver receiver = new ImapMailReceiver();
receiver.setShouldMarkMessagesAsRead(true);
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
when(folder.isOpen()).thenReturn(true);
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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();
@@ -64,13 +66,14 @@ public class ImapMailSearchTermsTests {
public void validateSearchTermsWhenShouldMarkAsReadWithExistingFlags() throws Exception {
ImapMailReceiver receiver = new ImapMailReceiver();
receiver.setShouldMarkMessagesAsRead(true);
receiver.afterPropertiesSet();
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
when(folder.isOpen()).thenReturn(true);
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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();
@@ -87,18 +90,19 @@ 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();
MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver);
Folder folder = context.getFolder();
when(folder.isOpen()).thenReturn(true);
Field folderField = AbstractMailReceiver.class.getDeclaredField("folder");
folderField.setAccessible(true);
Folder folder = mock(Folder.class);
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();

150
spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java
View File

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

36
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java
View File

@@ -17,7 +17,6 @@ package org.springframework.integration.mongodb.inbound;
import java.util.List;
import org.springframework.context.MessageSource;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
@@ -29,10 +28,12 @@ import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.PseudoTransactionalMessageSource;
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.util.ExpressionUtils;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -58,7 +59,7 @@ import com.mongodb.DBObject;
* @since 2.2
*/
public class MongoDbMessageSource extends IntegrationObjectSupport
implements PseudoTransactionalMessageSource<Object, MongoOperations>{
implements MessageSource<Object> {
private final Expression queryExpression;
@@ -215,29 +216,12 @@ public class MongoDbMessageSource extends IntegrationObjectSupport
.build();
}
Object holder = TransactionSynchronizationManager.getResource(this);
if (holder != null) {
Assert.isInstanceOf(MessageSourceResourceHolder.class, holder);
((MessageSourceResourceHolder) holder).addAttribute("mongoTemplate", this.mongoTemplate);
}
return message;
}
/**
* Returns the mongo-template.
*/
public MongoOperations getResource() {
return this.mongoTemplate;
}
public void afterCommit(Object object) {
}
public void afterRollback(Object object) {
}
public void afterReceiveNoTx(MongoOperations resource) {
}
public void afterSendNoTx(MongoOperations resource) {
}
}

7
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/inbound-adapter-config.xml
View File

@@ -63,9 +63,13 @@
auto-startup="false">
<int:poller fixed-rate="200" max-messages-per-poll="1">
<int:pseudo-transactional on-success-expression="@documentCleaner.remove(#resource, payload, headers.mongo_collectionName)" on-success-result-channel="nullChannel"/>
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-mongodb:inbound-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="@documentCleaner.remove(#mongoTemplate, payload, headers.mongo_collectionName)"/>
</int:transaction-synchronization-factory>
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapterWithConverter"
channel="replyChannel"
@@ -93,5 +97,6 @@
<int:queue/>
</int:channel>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
</beans>

21
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java
View File

@@ -29,9 +29,11 @@ import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.PseudoTransactionalMessageSource;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transaction.MessageSourceResourceHolder;
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
@@ -43,7 +45,7 @@ import org.springframework.util.Assert;
* @since 2.2
*/
public class RedisStoreMessageSource extends IntegrationObjectSupport
implements PseudoTransactionalMessageSource<RedisStore, RedisStore> {
implements MessageSource<RedisStore> {
private final ThreadLocal<RedisStore> resourceHolder = new ThreadLocal<RedisStore>();
@@ -117,7 +119,12 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
Assert.hasText(key, "Failed to determine the key for the collection");
RedisStore store = this.createStoreView(key);
this.resourceHolder.set(store);
Object holder = TransactionSynchronizationManager.getResource(this);
if (holder != null) {
Assert.isInstanceOf(MessageSourceResourceHolder.class, holder);
((MessageSourceResourceHolder) holder).addAttribute("store", store);
}
if (store instanceof Collection<?> && ((Collection<Object>)store).size() < 1){
return null;
@@ -158,12 +165,4 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
public void afterRollback(Object object) {
this.resourceHolder.remove();
}
public void afterReceiveNoTx(RedisStore resource) {
this.resourceHolder.remove();
}
public void afterSendNoTx(RedisStore resource) {
this.resourceHolder.remove();
}
}

22
spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd
View File

@@ -168,18 +168,6 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Channel to which Error Messages will be sent.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -193,6 +181,16 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Identifies channel that error messages will be sent to if a failure occurs in this
gateway's invocation. If no "error-channel" reference is provided, this gateway will
propagate Exceptions to the caller. To completely suppress Exceptions, provide a
reference to the "nullChannel" here.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

65
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java
View File

@@ -18,16 +18,24 @@ package org.springframework.integration.redis.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.support.collections.RedisList;
import org.springframework.data.redis.support.collections.RedisZSet;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
@@ -85,6 +93,42 @@ public class RedisCollectionsInboundChannelAdapterParserTests extends RedisAvail
context.close();
}
@Test
@RedisAvailable
public void testListInboundConfigurationWithSynchronizationAndRollback() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareList(jcf);
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<Object, Object>();
redisTemplate.setConnectionFactory(jcf);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
BoundListOperations<Object, Object> ops = redisTemplate.boundListOps("baz");
assertTrue(ops.size() == 0);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationAndRollback", SourcePollingChannelAdapter.class);
SubscribableChannel redisFailChannel = context.getBean("redisFailChannel", SubscribableChannel.class);
redisFailChannel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
throw new MessagingException("intentional");
}
});
spca.start();
Thread.sleep(1000);
ops = redisTemplate.boundListOps("baz");
assertTrue(ops.size() == 13);
ops = redisTemplate.boundListOps("bar");
assertTrue(ops.size() == 0);
spca.stop();
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
@@ -109,6 +153,27 @@ public class RedisCollectionsInboundChannelAdapterParserTests extends RedisAvail
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testListInboundConfigurationWithBeforeCommit() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareList(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationBeforeCommit", SourcePollingChannelAdapter.class);
spca.start();
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
QueueChannel adapterErrors = context.getBean("adapterErrors", QueueChannel.class);
Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertNotNull(adapterErrors.receive(2000));
spca.stop();
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")

4
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java
View File

@@ -54,7 +54,7 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests{
@Autowired @Qualifier("autoChannel.adapter")
private RedisInboundChannelAdapter autoChannelAdapter;
@Test
@Test
@RedisAvailable
public void validateConfiguration() {
RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class);
@@ -67,7 +67,7 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests{
assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
}
@Test
@Test
@RedisAvailable
public void testInboundChannelAdapterMessaging() throws Exception{
JedisConnectionFactory connectionFactory = new JedisConnectionFactory();

41
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml
View File

@@ -22,9 +22,21 @@
channel="redisChannel"
auto-startup="false">
<int:poller fixed-rate="2000" max-messages-per-poll="10">
<int:pseudo-transactional on-success-expression="#resource.rename('bar')"/>
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int-redis:store-inbound-channel-adapter id="listAdapterWithSynchronizationAndRollback"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="redisFailChannel"
auto-startup="false">
<int:poller fixed-rate="2000" max-messages-per-poll="10">
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int:channel id="redisFailChannel"/>
<int-redis:store-inbound-channel-adapter id="listAdapterWithSynchronizationAndRedisTemplate"
redis-template="redisTemplate"
@@ -32,9 +44,32 @@
channel="redisChannel"
auto-startup="false">
<int:poller fixed-rate="2000" max-messages-per-poll="10">
<int:pseudo-transactional on-success-expression="#resource.rename('bar')"/>
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="#resource.attributes['store'].rename('bar')"/>
<int:after-rollback expression="#store.rename('baz')"/>
</int:transaction-synchronization-factory>
<int-redis:store-inbound-channel-adapter id="listAdapterWithSynchronizationBeforeCommit"
redis-template="redisTemplate"
key-expression="'presidents'"
channel="redisChannel"
auto-startup="false">
<int:poller fixed-rate="2000" max-messages-per-poll="10" error-channel="adapterErrors">
<int:transactional synchronization-factory="syncFactory2"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int:channel id="adapterErrors">
<int:queue/>
</int:channel>
<int:transaction-synchronization-factory id="syncFactory2">
<int:before-commit expression="5/0"/>
</int:transaction-synchronization-factory>
<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
<property name="keySerializer">
@@ -53,5 +88,7 @@
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
<property name="port" value="7379"/>
</bean>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
</beans>

8
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml
View File

@@ -41,9 +41,13 @@
auto-startup="false"
collection-type="ZSET">
<int:poller fixed-rate="1000" max-messages-per-poll="2">
<int:pseudo-transactional on-success-expression="#resource.removeByScore(18, 18)" on-success-result-channel="nullChannel"/>
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="#resource.attributes['store'].removeByScore(18, 18)"/>
</int:transaction-synchronization-factory>
<int:channel id="redisChannel">
<int:queue/>
@@ -56,5 +60,7 @@
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
<property name="port" value="7379"/>
</bean>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
</beans>

13
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml
View File

@@ -50,13 +50,14 @@
comparator="comparator"
delete-remote-files="${delete.remote.files}">
<poller fixed-rate="1000">
<pseudo-transactional on-success-expression="'foo'"
on-success-result-channel="successChannel"
on-failure-expression="'bar'"
on-failure-result-channel="failureChannel"
send-timeout="123" />
<transactional synchronization-factory="syncFactory"/>
</poller>
</sftp:inbound-channel-adapter>
<transaction-synchronization-factory id="syncFactory">
<after-commit expression="'foo'" channel="successChannel"/>
<after-rollback expression="'bar'" channel="failureChannel"/>
</transaction-synchronization-factory>
<channel id="successChannel" />
@@ -114,5 +115,7 @@
</sftp:inbound-channel-adapter>
<bridge input-channel="autoChannel" output-channel="nullChannel" />
<beans:bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
</beans:beans>

12
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java
View File

@@ -29,11 +29,11 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
@@ -83,16 +83,6 @@ public class InboundChannelAdapterParserTests {
assertEquals(".", remoteFileSeparator);
PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class);
assertNotNull(requestChannel.receive(2000));
assertEquals("foo", TestUtils.getPropertyValue(adapter, "onSuccessExpression", Expression.class).getValue());
assertSame(context.getBean("successChannel"), TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "defaultChannel"));
assertEquals(123L, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onSuccessMessagingTemplate"), "sendTimeout"));
assertEquals("bar", TestUtils.getPropertyValue(adapter, "onFailureExpression", Expression.class).getValue());
assertSame(context.getBean("failureChannel"), TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "defaultChannel"));
assertEquals(123L, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(adapter, "onFailureMessagingTemplate"), "sendTimeout"));
}
@Test