From 4de02fa75e3feae3515e4bad1cd8451b20349c2a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 26 Jul 2012 18:23:49 -0400 Subject: [PATCH] INT-2685 Transaction Synchronization Remove M3 disposition-* attributes on File/(S)FTP inbound adapters. Add and elements to . These elements provide the following attributes: * on-success-expression * on-success-result-channel * on-failure-expression * on-failure-result-channel * send-timeout synchronizes these expression evaluations with the transaction, such that they are executed immediately after the commit/rollback. is used for a non-transactional poller. When an is provided to the poller, and are synonyms; and the behavior is dictated by whether or not the contains a transaction advice. It is recommended that is used when the does not have a txAdvice, and when it does, but the framework does not enforce this. The expressions have the original (polled) message as the #root variable. In addition, a BeanResolver is provided, allowing expressions such as '@someBean.handleSuccess(payload)'. MessageSources may also implement PseudoTransactionalMessageSource. This has a number of methods allowing more flexibility in transactional and non-transactional environents. For example, for backwards compatibility. the mail-inbound-channel-adapter deletes its polled message after the receive() rather than after the polled message is sent (when running in a non-transactional poller). However, when running in a transactional poller, the delete is done after the transaction commits (but not when it rolls back). In addition, MessageSources that implement this interface can optionally provide an arbitrary object to the success/failure expressions in a variable named '#resource'. INT-2685 Polishing PR Review Comments Add tests for non-tx PseudoTransactionalMessageSource --- .../integration/MessageHeaders.java | 14 +- ...ourcePollingChannelAdapterFactoryBean.java | 24 ++- .../integration/config/xml/PollerParser.java | 37 +++- .../PseudoTransactionalMessageSource.java | 16 +- .../endpoint/SourcePollingChannelAdapter.java | 192 ++++++++++++++--- .../scheduling/PollerMetadata.java | 58 +++++- .../config/xml/spring-integration-2.2.xsd | 114 ++++++++-- .../config/xml/PollerParserTests.java | 20 -- ...PseudoTransactionalMessageSourceTests.java | 156 ++++++++++++++ .../integration/file/FileHeaders.java | 2 - .../file/FileReadingMessageSource.java | 81 +------- ...RemoteFileInboundChannelAdapterParser.java | 1 - .../FileInboundChannelAdapterParser.java | 1 - .../file/config/FileNamespaceUtils.java | 43 ---- .../FileReadingMessageSourceFactoryBean.java | 29 --- ...InboundFileSynchronizingMessageSource.java | 39 +--- .../config/spring-integration-file-2.2.xsd | 36 ---- .../FileInboundTransactionTests-context.xml | 51 +++++ .../file/FileInboundTransactionTests.java | 194 ++++++++++++++++++ .../file/FileReadingMessageSourceTests.java | 16 -- .../FileToChannelIntegrationTests-context.xml | 7 +- .../file/FileToChannelIntegrationTests.java | 3 +- ...boundChannelAdapterParserTests-context.xml | 13 +- .../FileInboundChannelAdapterParserTests.java | 12 -- .../ftp/config/spring-integration-ftp-2.2.xsd | 36 ---- ...boundChannelAdapterParserTests-context.xml | 17 +- .../FtpInboundChannelAdapterParserTests.java | 15 +- ...ransactionalMessageSourceTests-context.xml | 7 +- ...PseudoTransactionalMessageSourceTests.java | 26 ++- .../mail/MailReceivingMessageSource.java | 12 +- .../config/spring-integration-sftp-2.2.xsd | 36 ---- ...boundChannelAdapterParserTests-context.xml | 27 ++- .../InboundChannelAdapterParserTests.java | 15 +- 33 files changed, 892 insertions(+), 458 deletions(-) delete mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java 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 e75bffff7b..170df014d8 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 @@ -34,7 +34,7 @@ import org.apache.commons.logging.LogFactory; /** * The headers for a {@link Message}.
- * IMPORTANT: MessageHeaders are immutable. Any mutating operation (e.g., put(..), putAll(..) etc.) + * IMPORTANT: MessageHeaders are immutable. Any mutating operation (e.g., put(..), putAll(..) etc.) * will result in {@link UnsupportedOperationException} * To create MessageHeaders instance use fluent MessageBuilder API * 
@@ -47,17 +47,18 @@ import org.apache.commons.logging.LogFactory;
  * headers.put("key2", "value2");
  * new GenericMessage("foo", headers);
  *
- * + * * @author Arjen Poutsma * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public final class MessageHeaders implements Map, Serializable { private static final long serialVersionUID = 6901029029524535147L; private static final Log logger = LogFactory.getLog(MessageHeaders.class); - + private static volatile IdGenerator idGenerator = null; /** @@ -88,6 +89,8 @@ public final class MessageHeaders implements Map, Serializable { public static final String CONTENT_TYPE = "content-type"; + public static final String DISPOSITION_RESULT = "dispositionResult"; + private final Map headers; @@ -100,7 +103,7 @@ public final class MessageHeaders implements Map, Serializable { else { this.headers.put(ID, MessageHeaders.idGenerator.generateId()); } - + this.headers.put(TIMESTAMP, new Long(System.currentTimeMillis())); } @@ -155,10 +158,12 @@ public final class MessageHeaders implements Map, Serializable { return (T) value; } + @Override public int hashCode() { return this.headers.hashCode(); } + @Override public boolean equals(Object obj) { if (this == obj) { return true; @@ -170,6 +175,7 @@ public final class MessageHeaders implements Map, Serializable { return false; } + @Override public String toString() { return this.headers.toString(); } 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 ee5521dfc7..538fcff1c2 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,6 +24,7 @@ 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; @@ -133,7 +134,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBeanAll {@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 * @@ -42,7 +46,9 @@ public interface PseudoTransactionalMessageSource extends MessageSource /** * Obtain the resource on which appropriate action needs - * to be taken. + * 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(); @@ -50,16 +56,16 @@ public interface PseudoTransactionalMessageSource extends MessageSource /** * Invoked via {@link TransactionSynchronization} when the * transaction commits. - * @param resource The resource to be "committed" + * @param object The resource to be "committed" */ - void afterCommit(V resource); + void afterCommit(Object object); /** * Invoked via {@link TransactionSynchronization} when the * transaction rolls back. - * @param resource + * @param object */ - void afterRollback(V resource); + void afterRollback(Object object); /** * Called when there is no transaction and the receive() call completed. 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 60a150fedd..598ef95d23 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,14 +16,21 @@ 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; @@ -40,6 +47,13 @@ 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; @@ -50,7 +64,15 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme private final MessagingTemplate messagingTemplate = new MessagingTemplate(); - private volatile boolean synchronizedTx = true; + private volatile Expression onSuccessExpression; + + private final MessagingTemplate onSuccessMessagingTemplate = new MessagingTemplate(); + + private volatile Expression onFailureExpression; + + private final MessagingTemplate onFailureMessagingTemplate = new MessagingTemplate(); + + private volatile StandardEvaluationContext evaluationContext; /** * Specify the source to be polled for Messages. @@ -82,8 +104,30 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme this.shouldTrack = shouldTrack; } - public void setSynchronized(boolean synchronizedTx) { - this.synchronizedTx = synchronizedTx; + + 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 @@ -96,6 +140,7 @@ 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(); } @@ -104,25 +149,32 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme protected boolean doPoll() { boolean isInTx = false; PseudoTransactionalMessageSource messageSource = null; - Object resource = null; + Object resource = NO_TX_RESOURCE; if (this.isPseudoTxMessageSource) { messageSource = (PseudoTransactionalMessageSource) this.source; resource = messageSource.getResource(); - Assert.state(resource != null, "Pseudo Transactional Message Source returned null resource"); - if (this.synchronizedTx && TransactionSynchronizationManager.isActualTransactionActive()) { - TransactionSynchronizationManager.bindResource(messageSource, resource); - TransactionSynchronizationManager.registerSynchronization( - new PseudoTransactionalResourceSynchronization( - new PseudoTransactionalResourceHolder(resource), this.source)); - isInTx = true; - } } Message message; try { message = this.source.receive(); + if (TransactionSynchronizationManager.isActualTransactionActive()) { + TransactionSynchronizationManager.bindResource(this, resource); + TransactionSynchronizationManager.registerSynchronization( + new PseudoTransactionalResourceSynchronization( + new PseudoTransactionalResourceHolder(message, resource), this)); + isInTx = true; + } } finally { - if (this.isPseudoTxMessageSource && !isInTx) { + 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); } } @@ -133,9 +185,30 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme if (this.shouldTrack) { message = MessageHistory.write(message, this); } - this.messagingTemplate.send(this.outputChannel, message); - if (this.isPseudoTxMessageSource && !isInTx) { - messageSource.afterSendNoTx(resource); + 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; + } + else { + throw new MessagingException(message, e); + } } return true; } @@ -145,11 +218,78 @@ 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(Object resource) { + public PseudoTransactionalResourceHolder(Message message, Object resource) { + this.message = message; this.resource = resource; } @@ -157,6 +297,10 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme return resource; } + public Message getMessage() { + return message; + } + public void reset() { } @@ -185,28 +329,30 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme return false; } - @SuppressWarnings("unchecked") @Override protected void processResourceAfterCommit(PseudoTransactionalResourceHolder resourceHolder) { if (logger.isTraceEnabled()) { logger.trace("'Committing' pseudo-transactional resource"); } - ((PseudoTransactionalMessageSource) source).afterCommit(resourceHolder.getResource()); + if (isPseudoTxMessageSource) { + ((PseudoTransactionalMessageSource) source).afterCommit(resourceHolder.getResource()); + } + onSuccess(resourceHolder.getMessage(), resourceHolder.getResource()); } - @SuppressWarnings("unchecked") @Override public void afterCompletion(int status) { if (status != TransactionSynchronization.STATUS_COMMITTED) { if (logger.isTraceEnabled()) { logger.trace("'Rolling back' pseudo-transactional resource"); } - ((PseudoTransactionalMessageSource) source).afterRollback(this.resourceHolder.getResource()); + 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 176b00bb82..bc5294741e 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,12 +20,15 @@ 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.scheduling.Trigger; import org.springframework.util.ErrorHandler; /** * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public class PollerMetadata { @@ -39,11 +42,19 @@ public class PollerMetadata { private volatile ErrorHandler errorHandler; - private List adviceChain; + private volatile List adviceChain; private volatile Executor taskExecutor; - private volatile boolean synchronizedTx = true; + private volatile Expression onSuccessExpression; + + private volatile MessageChannel onSuccessResultChannel; + + private volatile Expression onFailureExpression; + + private volatile MessageChannel onFailureResultChannel; + + private volatile long sendTimeout; public void setTrigger(Trigger trigger) { this.trigger = trigger; @@ -102,11 +113,44 @@ public class PollerMetadata { return this.taskExecutor; } - public boolean isSynchronized() { - return synchronizedTx; + public Expression getOnSuccessExpression() { + return onSuccessExpression; } - public void setSynchronized(boolean synchronizedTx) { - this.synchronizedTx = synchronizedTx; + 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; + } + + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + } 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 8c2efb737e..6e4f27d390 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 @@ -1504,6 +1504,34 @@ + + + + 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. + ]]> + + + @@ -1578,20 +1606,6 @@ ]]> - - - - Specifies whether the resource, used by the MessageSource that this poller - polls, is synchronized with the transaction. The resource may be disposed of - in different manners, depending on whether the transaction commits, - or rolls back. Only applied if a transaction subelement (or - an advice-chain that contains a transaction advice) is provided. - Also, only applies if the MessageSource implements - PseudoTransactionalMessageSource. - Default true. - - - @@ -3404,6 +3418,10 @@ is provided, the return value is expected to match a channel name exactly. + + + + @@ -3742,4 +3760,72 @@ endpoint itself is a Polling Consumer for a channel with a queue. + + + + or poller + sub element. + Used to take action after the transaction completes () or after + the channel.send() is complete (). + ]]> + + + + ), + or after the transaction commits (). The #root variable of the + expression evaluation is the original message; a BeanResolver is also available. + ]]> + + + + + + + + + + + + + + + ), + or after the transaction rolls back (). The #root variable of the + expression evaluation is the original message; a BeanResolver is also available. + ]]> + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java index 1192367fbc..38dd1ee712 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerParserTests.java @@ -140,24 +140,4 @@ public class PollerParserTests { "pollerWithCronAndFixedDelay.xml", PollerParserTests.class); } - @Test - public void pollerWithSync() { - ApplicationContext context = new ClassPathXmlApplicationContext( - "pollerWithSynchronization.xml", PollerParserTests.class); - Object poller = context.getBean("noSync"); - assertNotNull(poller); - PollerMetadata metadata = (PollerMetadata) poller; - assertEquals(true, metadata.isSynchronized()); - - poller = context.getBean("syncTrue"); - assertNotNull(poller); - metadata = (PollerMetadata) poller; - assertEquals(true, metadata.isSynchronized()); - - poller = context.getBean("syncFalse"); - assertNotNull(poller); - metadata = (PollerMetadata) poller; - assertEquals(false, metadata.isSynchronized()); - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java index 96a2fed6bf..d6f4830370 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 @@ -15,13 +15,17 @@ */ 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.message.GenericMessage; @@ -76,9 +80,88 @@ public class PseudoTransactionalMessageSourceTests { assertSame(object, committed.get()); 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(); @@ -118,7 +201,80 @@ public class PseudoTransactionalMessageSourceTests { TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); assertSame(object, rolledBack.get()); 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 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) { + } + }); + + 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); + + 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); + + 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)); + } + + public class Bar { + public String getValue() { + return "bar"; + } + } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java index b80561534d..3f4113501d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java @@ -35,6 +35,4 @@ public abstract class FileHeaders { public static final String REMOTE_FILE = PREFIX + "remoteFile"; - public static final String DISPOSITION_RESULT = PREFIX + "dispositionResult"; - } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index f24517bb76..f685015d88 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -26,22 +26,14 @@ import java.util.concurrent.PriorityBlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.context.expression.BeanFactoryResolver; -import org.springframework.expression.EvaluationContext; -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.MessagingException; import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.core.MessagingTemplate; -import org.springframework.integration.core.PseudoTransactionalMessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.util.ExpressionUtils; import org.springframework.util.Assert; /** @@ -73,7 +65,7 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gary Russell */ -public class FileReadingMessageSource extends IntegrationObjectSupport implements PseudoTransactionalMessageSource { +public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource { private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5; @@ -95,17 +87,8 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement private volatile boolean scanEachPoll = false; - private volatile Expression dispositionExpression; - - private final MessagingTemplate dispositionMessagingTemplate = new MessagingTemplate(); - - private volatile boolean dispostionResultChannelSet; - private final ThreadLocal resources = new ThreadLocal(); - private EvaluationContext evaluationContext = new StandardEvaluationContext(); - - /** * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. */ @@ -240,21 +223,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement this.scanEachPoll = scanEachPoll; } - public void setDispositionExpression(Expression dispositionExpression) { - Assert.notNull(dispositionExpression, "'dispositionExpression' must not be null"); - this.dispositionExpression = dispositionExpression; - } - - public void setDispositionResultChannel(MessageChannel dispositionResultChannel) { - Assert.notNull(dispositionResultChannel, "'dispositionResultChannel' must not be null"); - this.dispositionMessagingTemplate.setDefaultChannel(dispositionResultChannel); - this.dispostionResultChannelSet = true; - } - - public void setDispositionSendTimeout(long dispositionSendTimeout) { - this.dispositionMessagingTemplate.setSendTimeout(dispositionSendTimeout); - } - @Override public String getComponentType() { return "file:inbound-channel-adapter"; @@ -272,10 +240,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement "Source path [" + this.directory + "] does not point to a directory."); Assert.isTrue(this.directory.canRead(), "Source directory [" + this.directory + "] is not readable."); - if (getBeanFactory() != null) { - this.evaluationContext = ExpressionUtils.createStandardEvaluationContext( - new BeanFactoryResolver(getBeanFactory())); - } } public Message receive() throws MessagingException { @@ -345,47 +309,4 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } } - public FileMessageHolder getResource() { - FileMessageHolder resource = new FileMessageHolder(); - this.resources.set(resource); - return resource; - } - - public void afterCommit(FileMessageHolder resource) { - Assert.isInstanceOf(FileMessageHolder.class, resource); - FileMessageHolder fileResource = resource; - if (this.dispositionExpression != null) { - if (logger.isDebugEnabled()) { - logger.debug("Executing expression " + this.dispositionExpression.getExpressionString() + " on " + - fileResource.getMessage()); - } - Object result = this.dispositionExpression.getValue(this.evaluationContext, fileResource.getMessage()); - if (result != null) { - if (this.dispostionResultChannelSet) { - try { - Message message = MessageBuilder.fromMessage(fileResource.getMessage()) - .setHeader(FileHeaders.DISPOSITION_RESULT, result).build(); - this.dispositionMessagingTemplate.send(message); - } - catch (Exception e) { - logger.error("Error sending File Disposition Result", e); - } - } - } - } - this.resources.set(null); - } - - public void afterRollback(FileMessageHolder resource) { - // no op - } - - public void afterReceiveNoTx(FileMessageHolder resource) { - // no op - } - - public void afterSendNoTx(FileMessageHolder resource) { - this.afterCommit(resource); - } - } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java index cebcac8d0f..0de8aebcc0 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java @@ -70,7 +70,6 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst localFileGeneratorExpressionBuilder.addConstructorArgValue(localFileGeneratorExpression); synchronizerBuilder.addPropertyValue("localFilenameGeneratorExpression", localFileGeneratorExpressionBuilder.getBeanDefinition()); } - FileNamespaceUtils.setDispositionAttributes(element, messageSourceBuilder); return messageSourceBuilder.getBeanDefinition(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java index a3fcebcce1..d6261d48de 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java @@ -47,7 +47,6 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size"); - FileNamespaceUtils.setDispositionAttributes(element, builder); String filterBeanName = this.registerFilter(element, parserContext); String lockerBeanName = registerLocker(element, parserContext); if (lockerBeanName != null) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java deleted file mode 100644 index 186de34831..0000000000 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java +++ /dev/null @@ -1,43 +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.file.config; - -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.integration.config.ExpressionFactoryBean; -import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.springframework.util.StringUtils; -import org.w3c.dom.Element; - -/** - * @author Gary Russell - * @since 2.2 - * - */ -public class FileNamespaceUtils { - - public static void setDispositionAttributes(Element element, BeanDefinitionBuilder builder) { - String dispositionExpression = element.getAttribute("disposition-expression"); - if (StringUtils.hasText(dispositionExpression)) { - RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class); - expressionDef.getConstructorArgumentValues().addGenericArgumentValue(dispositionExpression); - builder.addPropertyValue("dispositionExpression", expressionDef); - } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "disposition-result-channel"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "disposition-send-timeout"); - } - -} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java index a0514069c7..b1d7ca2eab 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java @@ -22,8 +22,6 @@ import java.util.Comparator; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.FactoryBean; -import org.springframework.expression.Expression; -import org.springframework.integration.MessageChannel; import org.springframework.integration.file.DirectoryScanner; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.filters.CompositeFileListFilter; @@ -57,12 +55,6 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean extends MessageProducerSupport - implements PseudoTransactionalMessageSource { + implements MessageSource { /** * Should the endpoint attempt to create the local directory? True by default. @@ -107,18 +104,6 @@ public abstract class AbstractInboundFileSynchronizingMessageSource extends M this.localDirectory = localDirectory; } - public void setDispositionExpression(Expression dispositionExpression) { - this.fileSource.setDispositionExpression(dispositionExpression); - } - - public void setDispositionResultChannel(MessageChannel dispositionResultChannel) { - this.fileSource.setDispositionResultChannel(dispositionResultChannel); - } - - public void setDispositionSendTimeout(long dispositionSendTimeout) { - this.fileSource.setDispositionSendTimeout(dispositionSendTimeout); - } - @Override protected void onInit() { Assert.notNull(this.localDirectory, "localDirectory must not be null"); @@ -172,24 +157,4 @@ public abstract class AbstractInboundFileSynchronizingMessageSource extends M new RegexPatternFileListFilter(completePattern))); } - public FileMessageHolder getResource() { - return this.fileSource.getResource(); - } - - public void afterCommit(FileMessageHolder resource) { - this.fileSource.afterCommit(resource); - } - - public void afterRollback(FileMessageHolder resource) { - this.fileSource.afterRollback(resource); - } - - public void afterReceiveNoTx(FileMessageHolder resource) { - this.fileSource.afterReceiveNoTx(resource); - } - - public void afterSendNoTx(FileMessageHolder resource) { - this.fileSource.afterSendNoTx(resource); - } - } diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.2.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.2.xsd index 24628294f9..4fbca60514 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.2.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.2.xsd @@ -166,42 +166,6 @@ Only files matching this regular expression will be picked up by this adapter. - - - - SpEL expression to be executed after the message has been sent. If running in a transactional - poller, it will be executed after the transaction commits. If running in a non-transactional - poller it will execute after the message is sent. Note that the actual point of execution - depends on any asynchronous handoffs on the downstream flow. It will be executed when the - current thread returns from the channel send. The root object of the expression is the - original message (with a File payload). Examples: "payload.delete()", - "payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)". - - - - - - - If a 'disposition-expression' is provided, and that expression returns a result, the result - is sent to this channel, with the original message payload and a 'file_dispositionResult' - header containing the result of the expression execution. - - - - - - - - - - - - If a 'disposition-expression' is provided, and that expression returns a result, the result - is sent to this disposition-result-channel. This timeout specifies how long to wait if - that channel might block (such as a bounded queue channel that is full). Default infinity. - - - 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 new file mode 100644 index 0000000000..6170bef8a3 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml @@ -0,0 +1,51 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..d125174175 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java @@ -0,0 +1,194 @@ +/* + * 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.file; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.util.concurrent.CountDownLatch; +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; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionException; +import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionStatus; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class FileInboundTransactionTests { + + @Autowired + private SourcePollingChannelAdapter pseudoTx; + + @Autowired + private SourcePollingChannelAdapter realTx; + + @Autowired + private SubscribableChannel input; + + @Autowired + private SubscribableChannel txInput; + + @Autowired + private PollableChannel successChannel; + + @Autowired + private PollableChannel failureChannel; + + @Autowired + private DummyTxManager transactionManager; + + @Value("${java.io.tmpdir}") + private String tmpDir; + + @Test + public void testNoTx() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean crash = new AtomicBoolean(); + input.subscribe(new MessageHandler() { + + public void handleMessage(Message message) throws MessagingException { + System.out.println(message); + if (crash.get()) { + throw new MessagingException("eek"); + } + latch.countDown(); + } + }); + pseudoTx.start(); + new File(tmpDir + "/si-test1").mkdir(); + File file = new File(tmpDir + "/si-test1/foo"); + file.createNewFile(); + Message result = successChannel.receive(10000); + assertNotNull(result); + assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT)); + System.out.println(result); + assertFalse(file.delete()); + crash.set(true); + file = new File(tmpDir + "/si-test1/bar"); + file.createNewFile(); + result = failureChannel.receive(10000); + assertNotNull(result); + System.out.println(result); + assertTrue(file.delete()); + assertEquals("foo", result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT)); + pseudoTx.stop(); + assertFalse(transactionManager.getCommitted()); + assertFalse(transactionManager.getRolledBack()); + } + + @Test + public void testTx() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean crash = new AtomicBoolean(); + txInput.subscribe(new MessageHandler() { + + public void handleMessage(Message message) throws MessagingException { + System.out.println(message); + if (crash.get()) { + throw new MessagingException("eek"); + } + latch.countDown(); + } + }); + realTx.start(); + new File(tmpDir + "/si-test2").mkdir(); + File file = new File(tmpDir + "/si-test2/baz"); + file.createNewFile(); + Message result = successChannel.receive(10000); + assertNotNull(result); + assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT)); + assertTrue(file.delete()); + System.out.println(result); + assertTrue(transactionManager.getCommitted()); + crash.set(true); + file = new File(tmpDir + "/si-test2/qux"); + file.createNewFile(); + result = failureChannel.receive(10000); + assertNotNull(result); + System.out.println(result); + assertTrue(file.delete()); + assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT)); + realTx.stop(); + assertTrue(transactionManager.getRolledBack()); + } + + public static class DummyTxManager extends AbstractPlatformTransactionManager { + + private static final long serialVersionUID = 1L; + + boolean committed; + + boolean rolledBack; + + @Override + protected Object doGetTransaction() throws TransactionException { + return new Object(); + } + + @Override + protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException { + } + + @Override + protected void doCommit(DefaultTransactionStatus status) throws TransactionException { + committed = true; + } + + @Override + protected void doRollback(DefaultTransactionStatus status) throws TransactionException { + rolledBack = true; + } + + /** + * Evaluated in transactional onSuccessExpression - ensures we rolled back before evaluation + * @return + */ + public boolean getCommitted() { + return committed; + } + + /** + * Evaluated in transactional onFailureExpression - ensures we rolled back before evaluation + * @return + */ + public boolean getRolledBack() { + return rolledBack; + } + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index 64fcfb4907..c63c9d9f50 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java @@ -37,10 +37,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; -import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.Message; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.message.GenericMessage; /** * @author Iwein Fuld @@ -159,17 +156,4 @@ public class FileReadingMessageSourceTests { verify(inputDirectoryMock, times(2)).listFiles(); } - @Test - public void disposition() { - source.setDispositionExpression(new LiteralExpression("foo")); - QueueChannel channel = new QueueChannel(); - source.setDispositionResultChannel(channel); - FileMessageHolder resource = source.getResource(); - File file = mock(File.class); - resource.setMessage(new GenericMessage(file)); - source.afterCommit(resource); - Message result = channel.receive(10000); - assertSame(file, result.getPayload()); - assertEquals("foo", result.getHeaders().get(FileHeaders.DISPOSITION_RESULT)); - } } 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 7e76e7bd6f..c461a74901 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 @@ -12,8 +12,6 @@ - + + + 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 7cf6972260..345ca9c3d5 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 @@ -27,6 +27,7 @@ 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; @@ -58,7 +59,7 @@ public class FileToChannelIntegrationTests { assertNotNull(received.getPayload()); Message result = resultChannel.receive(10000); assertNotNull(result); - assertEquals(Boolean.TRUE, result.getHeaders().get(FileHeaders.DISPOSITION_RESULT)); + assertEquals(Boolean.TRUE, result.getHeaders().get(MessageHeaders.DISPOSITION_RESULT)); 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 074fd6cb78..293d89fec9 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 @@ -14,14 +14,17 @@ directory="${java.io.tmpdir}" filter="filter" comparator="testComparator" - disposition-expression="payload.delete()" - disposition-result-channel="resultChannel" - disposition-send-timeout="123" auto-startup="false"> - + + + - + 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 70334a2a9a..9728de1f60 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 @@ -30,13 +30,10 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; -import org.springframework.expression.Expression; -import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.file.DefaultDirectoryScanner; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; -import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -94,15 +91,6 @@ public class FileInboundChannelAdapterParserTests { assertSame("comparator reference not set, ", expected, actual); } - @Test - public void disposition() throws Exception { - Object dispositionExpression = accessor.getPropertyValue("dispositionExpression"); - assertEquals(SpelExpression.class, dispositionExpression.getClass()); - assertEquals("payload.delete()", ((Expression) dispositionExpression).getExpressionString()); - assertSame(TestUtils.getPropertyValue(source, "dispositionMessagingTemplate.defaultChannel"), context.getBean("resultChannel")); - assertEquals(123L, TestUtils.getPropertyValue(source, "dispositionMessagingTemplate.sendTimeout")); - } - static class TestComparator implements Comparator { public int compare(File f1, File f2) { diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd index 15e45eaedd..24e064c1d7 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd @@ -218,42 +218,6 @@ - - - - SpEL expression to be executed after the message has been sent. If running in a transactional - poller, it will be executed after the transaction commits. If running in a non-transactional - poller it will execute after the message is sent. Note that the actual point of execution - depends on any asynchronous handoffs on the downstream flow. It will be executed when the - current thread returns from the channel send. The root object of the expression is the - original message (with a File payload). Examples: "payload.delete()", - "payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)". - - - - - - - If a 'disposition-expression' is provided, and that expression returns a result, the result - is sent to this channel, with the original message payload and a 'file_dispositionResult' - header containing the result of the expression execution. - - - - - - - - - - - - If a 'disposition-expression' is provided, and that expression returns a result, the result - is sent to this disposition-result-channel. This timeout specifies how long to wait if - that channel might block (such as a bounded queue channel that is full). Default infinity. - - - 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 c817e8d4cd..e1eda62862 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 @@ -24,14 +24,19 @@ local-filename-generator-expression="#this.toUpperCase() + '.a'" comparator="comparator" temporary-file-suffix=".foo" - remote-directory="foo/bar" - disposition-expression="'foo'" - disposition-result-channel="dispoChannel" - disposition-send-timeout="123"> - + remote-directory="foo/bar"> + + + - + + + 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 76b885578d..328a7cee97 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 @@ -35,7 +35,6 @@ 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.FileReadingMessageSource; import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; @@ -78,12 +77,16 @@ public class FtpInboundChannelAdapterParserTests { assertNotNull(filter); Object sessionFactory = TestUtils.getPropertyValue(fisync, "sessionFactory"); assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass())); - FileReadingMessageSource source = TestUtils.getPropertyValue(inbound, "fileSource", FileReadingMessageSource.class); - assertEquals("foo", TestUtils.getPropertyValue(source, "dispositionExpression", Expression.class).getValue()); - assertSame(ac.getBean("dispoChannel"), TestUtils.getPropertyValue( - TestUtils.getPropertyValue(source, "dispositionMessagingTemplate"), "defaultChannel")); + 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(source, "dispositionMessagingTemplate"), "sendTimeout")); + 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 index 8aff133da4..337582d4d5 100644 --- 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 @@ -9,16 +9,15 @@ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd"> - + + - - - + 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 index 8c0501dcf6..269ec66c78 100644 --- 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 @@ -22,8 +22,12 @@ 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; @@ -45,27 +49,39 @@ public class PseudoTransactionalMessageSourceTests { private static boolean rolledBack; - private static boolean doRollback; + @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 { - doRollback = true; + 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() { - if (doRollback) { - throw new RuntimeException("Expected"); - } return new GenericMessage("foo"); } 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 c5376bfcd8..91e44b7664 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 @@ -81,25 +81,25 @@ public class MailReceivingMessageSource implements PseudoTransactionalMessageSou return this.mailReceiver.getTransactionContext(); } - public void afterCommit(MailReceiverContext context) { + public void afterCommit(Object context) { Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext"); - this.mailReceiver.closeContextAfterSuccess(context); + this.mailReceiver.closeContextAfterSuccess((MailReceiverContext) context); } - public void afterRollback(MailReceiverContext context) { + public void afterRollback(Object context) { Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext"); - this.mailReceiver.closeContextAfterFailure(context); + this.mailReceiver.closeContextAfterFailure((MailReceiverContext) context); } /** - * For backwards-compatibility; the mail adapter updates the status before the send. + * 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; the mail adapter updates the status before the send. + * 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-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd index 5aa3f7d624..ebccdbca7a 100644 --- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd @@ -221,42 +221,6 @@ - - - - SpEL expression to be executed after the message has been sent. If running in a transactional - poller, it will be executed after the transaction commits. If running in a non-transactional - poller it will execute after the message is sent. Note that the actual point of execution - depends on any asynchronous handoffs on the downstream flow. It will be executed when the - current thread returns from the channel send. The root object of the expression is the - original message (with a File payload). Examples: "payload.delete()", - "payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)". - - - - - - - If a 'disposition-expression' is provided, and that expression returns a result, the result - is sent to this channel, with the original message payload and a 'file_dispositionResult' - header containing the result of the expression execution. - - - - - - - - - - - - If a 'disposition-expression' is provided, and that expression returns a result, the result - is sent to this disposition-result-channel. This timeout specifies how long to wait if - that channel might block (such as a bounded queue channel that is full). Default infinity. - - - 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 7332dcd96b..5c4d95ee58 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 @@ -48,19 +48,24 @@ local-filename-generator-expression="#this.toUpperCase() + '.a'" temporary-file-suffix=".bar" comparator="comparator" - delete-remote-files="${delete.remote.files}" - disposition-expression="'foo'" - disposition-result-channel="dispoChannel" - disposition-send-timeout="123"> - + delete-remote-files="${delete.remote.files}"> + + + - - + + + + - + - + - + - + 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 055d1cf794..a856b3bf5f 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 @@ -37,7 +37,6 @@ import org.springframework.expression.Expression; import org.springframework.integration.MessageChannel; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; -import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer; import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource; import org.springframework.integration.test.util.TestUtils; @@ -84,12 +83,16 @@ public class InboundChannelAdapterParserTests { assertEquals(".", remoteFileSeparator); PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class); assertNotNull(requestChannel.receive(2000)); - FileReadingMessageSource fileSource = TestUtils.getPropertyValue(source, "fileSource", FileReadingMessageSource.class); - assertEquals("foo", TestUtils.getPropertyValue(fileSource, "dispositionExpression", Expression.class).getValue()); - assertSame(context.getBean("dispoChannel"), TestUtils.getPropertyValue( - TestUtils.getPropertyValue(fileSource, "dispositionMessagingTemplate"), "defaultChannel")); + 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(fileSource, "dispositionMessagingTemplate"), "sendTimeout")); + 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