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