diff --git a/build.gradle b/build.gradle index aa071448d6..0d3481f55e 100644 --- a/build.gradle +++ b/build.gradle @@ -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"', diff --git a/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java b/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java index d6d38e18de..4be5a9617a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java @@ -89,8 +89,6 @@ public final class MessageHeaders implements Map, Serializable { public static final String CONTENT_TYPE = "content-type"; - public static final String DISPOSITION_RESULT = "dispositionResult"; - public static final String POSTPROCESS_RESULT = "postProcessResult"; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index 538fcff1c2..3bbebc8a3d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -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 - * For example, with a MailReceivingMessageSource, the email can be deleted - * on successful commit, but not deleted if the transaction rolls back. - *

- * This implements the 'Best Chance 1PC' pattern where there is only a - * small (but present) window in which a transaction might commit but the - * resource is not updated to reflect that. This could result in - * duplicate messages. - *

All {@link MessageSource}s can have success/failure expressions evaluated either as part - * of a transaction with a <transactional/> poller or after success/failure when - * running in a <pseudo-transactional/> 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 extends MessageSource { - - /** - * 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); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 801945db81..766d2397cb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -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 messageSource = null; - Object resource = NO_TX_RESOURCE; - if (this.isPseudoTxMessageSource) { - messageSource = (PseudoTransactionalMessageSource) 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 { - - 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); - } - - } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java index bc5294741e..9a4ae731f4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java @@ -20,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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java new file mode 100644 index 0000000000..fe949e4755 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java @@ -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 { + + 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); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java new file mode 100644 index 0000000000..72c3697ad4 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java @@ -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 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()); + } + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/MessageSourceResourceHolder.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/MessageSourceResourceHolder.java new file mode 100644 index 0000000000..b33508ab43 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/MessageSourceResourceHolder.java @@ -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 attributes = new HashMap(); + + 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 getAttributes() { + return Collections.unmodifiableMap(attributes); + } + + public void reset() { + } + + public void unbound() { + } + + public boolean isVoid() { + return false; + } + +} \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/PseudoTransactionManager.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/PseudoTransactionManager.java new file mode 100644 index 0000000000..41939f2733 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/PseudoTransactionManager.java @@ -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 not make such + * sources transactional; rather, together with the 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 + } +} \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactory.java new file mode 100644 index 0000000000..abfeae642e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactory.java @@ -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); +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java new file mode 100644 index 0000000000..43c999e1ba --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationProcessor.java @@ -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); + +} \ No newline at end of file diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd index c0c0d00e44..509bf38a44 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd @@ -34,6 +34,66 @@ + + + + + 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. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -1494,34 +1554,6 @@ - - - - element is present. - ]]> - - - - - 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 or element is present. - ]]> - - - @@ -3417,6 +3449,19 @@ is provided, the return value is expected to match a channel name exactly. ]]> + + + + + + + + + + @@ -3442,6 +3487,20 @@ is provided, the return value is expected to match a channel name exactly. + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-config.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-config.xml new file mode 100644 index 0000000000..9f63b479ea --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-config.xml @@ -0,0 +1,17 @@ + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-xsd.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-xsd.xml new file mode 100644 index 0000000000..468b00cd7e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests-xsd.xml @@ -0,0 +1,41 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests.java new file mode 100644 index 0000000000..b2f9d4558f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/TransactionSynchronizationFactoryParserTests.java @@ -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()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java index d6f4830370..a7d906cbc6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java @@ -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 committed = new AtomicReference(); - final AtomicReference rolledBack = new AtomicReference(); - adapter.setSource(new PseudoTransactionalMessageSource() { + adapter.setSource(new MessageSource() { public Message receive() { - return new GenericMessage("foo"); - } - - public Object getResource() { - return object; - } - - public void afterCommit(Object resource) { - committed.set(resource); - } - - public void afterRollback(Object resource) { - rolledBack.set(resource); - } - - public void afterReceiveNoTx(Object resource) { - } - - public void afterSendNoTx(Object resource) { + GenericMessage message = new GenericMessage("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 afterReceive = new AtomicReference(); - final AtomicReference afterSend = new AtomicReference(); - adapter.setSource(new PseudoTransactionalMessageSource() { - - public Message receive() { - return new GenericMessage("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 afterReceive = new AtomicReference(); - adapter.setSource(new PseudoTransactionalMessageSource() { - - public Message 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 committed = new AtomicReference(); - final AtomicReference rolledBack = new AtomicReference(); - adapter.setSource(new PseudoTransactionalMessageSource() { + adapter.setSource(new MessageSource() { public Message receive() { - return new GenericMessage("foo"); - } - - public Object getResource() { - return object; - } - - public void afterCommit(Object resource) { - committed.set(resource); - } - - public void afterRollback(Object resource) { - rolledBack.set(resource); - } - - public void afterReceiveNoTx(Object resource) { - } - - public void afterSendNoTx(Object resource) { + GenericMessage message = new GenericMessage("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 committed = new AtomicReference(); - final AtomicReference rolledBack = new AtomicReference(); - adapter.setSource(new PseudoTransactionalMessageSource() { + public void testCommitWithManager() { + final PollableChannel queueChannel = new QueueChannel(); + TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager()); + transactionTemplate.execute(new TransactionCallback() { - public Message receive() { - return new GenericMessage("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() { - public void afterReceiveNoTx(Object resource) { - } + public Message receive() { + GenericMessage message = new GenericMessage("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() { - 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() { + + public Message receive() { + GenericMessage message = new GenericMessage("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() { + + 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() { + + public Message receive() { + GenericMessage message = new GenericMessage("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 { diff --git a/spring-integration-file/.springBeans b/spring-integration-file/.springBeans index 211af10331..97f972fc8d 100644 --- a/spring-integration-file/.springBeans +++ b/spring-integration-file/.springBeans @@ -1,13 +1,14 @@ - - - 1 - - - - - - - - - - + + + 1 + + + + + + + src/test/java/org/springframework/integration/file/config/config.xml + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml index 1dbfb9e29b..d1d2dbf365 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml @@ -14,10 +14,7 @@ - + @@ -36,15 +33,34 @@ - - + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java index d125174175..a2115fecb0 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java @@ -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()); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml index 8dc42e3ca9..ed00347878 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml @@ -36,8 +36,7 @@ - + @@ -45,4 +44,15 @@ + + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java index 345ca9c3d5..46c18ed6b6 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java @@ -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()); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml index eb1b822631..6ca5a721d0 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml @@ -16,11 +16,7 @@ comparator="testComparator" auto-startup="false"> - + @@ -38,5 +34,18 @@ + + + + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java index 9728de1f60..8d9ad0c046 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java @@ -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; diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml index 67f2656c39..96a5762125 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml @@ -26,13 +26,14 @@ temporary-file-suffix=".foo" remote-directory="foo/bar"> - + + + + + + @@ -78,5 +79,7 @@ + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index 328a7cee97..1d1da1e8e9 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java @@ -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 diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests-context.xml deleted file mode 100644 index 337582d4d5..0000000000 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests-context.xml +++ /dev/null @@ -1,27 +0,0 @@ - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java deleted file mode 100644 index 269ec66c78..0000000000 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java +++ /dev/null @@ -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 { - - public Message receive() { - return new GenericMessage("foo"); - } - - public Object getResource() { - return new Object(); - } - - public void afterCommit(Object resource) { - committed = true; - latch1.countDown(); - } - - public void afterRollback(Object resource) { - rolledBack = true; - latch2.countDown(); - } - - public void afterReceiveNoTx(Object resource) { - } - - public void afterSendNoTx(Object resource) { - } - - } -} diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java index 06581db583..52d4c297df 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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 contextHolder = new ThreadLocal(); + 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 copiedMessages = new LinkedList(); - 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 copiedMessages = new LinkedList(); + logger.debug("Recieved " + messages.length + " messages"); + + boolean recentFlagSupported = false; + + Flags flags = this.getFolder().getPermanentFlags(); + + if (flags != null){ + recentFlagSupported = flags.contains(Flags.Flag.RECENT); + } + + for (int i = 0; i < messages.length; i++) { + if (!recentFlagSupported){ + if (flags != null && flags.contains(Flags.Flag.USER)){ + if (logger.isDebugEnabled()){ + logger.debug("USER flags are supported by this mail server. Flagging message with '" + SI_USER_FLAG + "' user flag"); + } + Flags siFlags = new Flags(); + siFlags.add(SI_USER_FLAG); + messages[i].setFlags(siFlags, true); + } + else { + if (logger.isDebugEnabled()){ + logger.debug("USER flags are not supported by this mail server. Flagging message with system flag"); + } + messages[i].setFlag(Flags.Flag.FLAGGED, true); + } + } + if (this.selectorExpression != null) { + Message message = messages[i]; + if (this.selectorExpression.getValue(this.context, message, Boolean.class)){ + this.setAdditionalFlags(message); + copiedMessages.add(new MimeMessage((MimeMessage) message)); + } + else { + if (logger.isDebugEnabled()){ + logger.debug("Fetched email with subject '" + message.getSubject() + "' will be discarded by the matching filter" + + " and will not be flagged as SEEN."); + } + } + } + else { + this.setAdditionalFlags(messages[i]); + copiedMessages.add(new MimeMessage((MimeMessage) messages[i])); + } + } + if (this.shouldDeleteMessages()) { + this.deleteMessages(messages); + } + return copiedMessages.toArray(new Message[copiedMessages.size()]); } - 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(); - } - } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java index fae74bc06d..f7150bdcf5 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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; } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceiver.java index 161ff0b898..b06ad801bb 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceiver.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceiver.java @@ -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; } - } } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java index 91e44b7664..f1210f9062 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java @@ -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 { +public class MailReceivingMessageSource implements MessageSource { private final Log logger = LogFactory.getLog(this.getClass()); @@ -53,7 +52,6 @@ public class MailReceivingMessageSource implements PseudoTransactionalMessageSou this.mailReceiver = mailReceiver; } - public 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 - } } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java index fcbda40c6f..962518e4b0 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return true; } }).when(folder).isOpen(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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")); - } - } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java index b94c9dc629..0031219c50 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailSearchTermsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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(); diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java index 6938b3841d..cf62e56d91 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/Pop3MailReceiverTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { 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() { - public Object answer(InvocationOnMock invocation) throws Throwable { - return null; - } - }).when(receiver).openFolder(); - adapter.setSource(new MailReceivingMessageSource(receiver)); - - TransactionSynchronizationManager.initSynchronization(); - TransactionSynchronizationManager.setActualTransactionActive(true); - final AtomicReference doPollMethod = new AtomicReference(); - ReflectionUtils.doWithMethods(SourcePollingChannelAdapter.class, new MethodCallback() { - - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - if (method.getName() == "doPoll") { - doPollMethod.set(method); - method.setAccessible(true); - } - } - }); - doPollMethod.get().invoke(adapter, (Object[]) null); - TransactionSynchronizationUtils.triggerAfterCommit(); - TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED); - TransactionSynchronizationManager.clearSynchronization(); - verify(folder).close(true); - } - - @Test - public void testRollback() throws Exception { - SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); - QueueChannel outputChannel = new QueueChannel(); - adapter.setOutputChannel(outputChannel); - Pop3MailReceiver receiver = new Pop3MailReceiver("pop3://some.host"); - receiver.setShouldDeleteMessages(true); - receiver = spy(receiver); - MailReceiverContext context = MailTestsHelper.setupContextHolder(receiver); - Folder folder = context.getFolder(); - when(folder.isOpen()).thenReturn(true); - doAnswer(new Answer() { - public Object answer(InvocationOnMock invocation) throws Throwable { - return null; - } - }).when(receiver).openFolder(); - adapter.setSource(new MailReceivingMessageSource(receiver)); - - TransactionSynchronizationManager.initSynchronization(); - TransactionSynchronizationManager.setActualTransactionActive(true); - final AtomicReference doPollMethod = new AtomicReference(); - ReflectionUtils.doWithMethods(SourcePollingChannelAdapter.class, new MethodCallback() { - - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - if (method.getName() == "doPoll") { - doPollMethod.set(method); - method.setAccessible(true); - } - } - }); - doPollMethod.get().invoke(adapter, (Object[]) null); - TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); - TransactionSynchronizationManager.clearSynchronization(); - verify(folder).close(false); - } - } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java index d07ae89bc2..390a5d2802 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java @@ -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{ + implements MessageSource { 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) { - - } } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/inbound-adapter-config.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/inbound-adapter-config.xml index 1af2bbf0df..1f21852b36 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/inbound-adapter-config.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/inbound-adapter-config.xml @@ -63,9 +63,13 @@ auto-startup="false"> - + + + + + + diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java index a299ed6b2c..daee63d903 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java @@ -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 { + implements MessageSource { private final ThreadLocal resourceHolder = new ThreadLocal(); @@ -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)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(); - } } diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd index 0fc23aef63..2b3b7799b0 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd @@ -168,18 +168,6 @@ - - - - - Channel to which Error Messages will be sent. - - - - - - - + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java index a5af6a4e07..e871581f98 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java @@ -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 redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(jcf); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + + BoundListOperations 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> message = (Message>) redisChannel.receive(1000); + assertNotNull(message); + assertNotNull(adapterErrors.receive(2000)); + spca.stop(); + context.close(); + } + + @Test @RedisAvailable @SuppressWarnings("unchecked") diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java index 12ba56ce3b..0a6e39811a 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java @@ -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(); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml index bae69471ce..68a4577233 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml @@ -22,9 +22,21 @@ channel="redisChannel" auto-startup="false"> - + + + + + + + + + - + + + + + + + + + + + + + + + + + + + + @@ -53,5 +88,7 @@ + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml index a76853a741..ecdc30682a 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml @@ -41,9 +41,13 @@ auto-startup="false" collection-type="ZSET"> - + + + + + @@ -56,5 +60,7 @@ + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml index d2402bf54d..8e4a85fae8 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml @@ -50,13 +50,14 @@ comparator="comparator" delete-remote-files="${delete.remote.files}"> - + + + + + + @@ -114,5 +115,7 @@ + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java index a856b3bf5f..1a99794e6c 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java @@ -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