diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java index b2b257adb9..a5d93e4799 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java @@ -22,6 +22,7 @@ import org.springframework.beans.factory.config.BeanDefinitionHolder; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.ConfigurationException; @@ -46,8 +47,12 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio private static final String POLLER_ELEMENT = "poller"; + private static final String SELECTOR_ATTRIBUTE = "selector"; + private static final String ERROR_HANDLER_ATTRIBUTE = "error-handler"; + private static final String INTERCEPTORS_ELEMENT = "interceptors"; + @Override protected Class getBeanClass(Element element) { @@ -92,7 +97,14 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio } IntegrationNamespaceUtils.setReferenceIfAttributeDefined( builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, SELECTOR_ATTRIBUTE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, ERROR_HANDLER_ATTRIBUTE); + Element interceptorsElement = DomUtils.getChildElementByTagName(element, INTERCEPTORS_ELEMENT); + if (interceptorsElement != null) { + EndpointInterceptorParser parser = new EndpointInterceptorParser(); + ManagedList interceptors = parser.parseInterceptors(interceptorsElement, parserContext); + builder.addPropertyValue("interceptors", interceptors); + } } private String parseAdapter(String ref, String method, Element element, ParserContext parserContext) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java index 514a64a26a..9bd8add796 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java @@ -16,19 +16,25 @@ package org.springframework.integration.config; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.DefaultMessageHandler; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.ServiceActivatorEndpoint; +import org.springframework.integration.message.MessageMappingMethodInvoker; /** * Parser for the <service-activator> element. * * @author Mark Fisher */ -public class ServiceActivatorParser extends AbstractMessageEndpointParser { +public class ServiceActivatorParser extends AbstractEndpointParser { @Override - protected Class getHandlerAdapterClass() { - return DefaultMessageHandler.class; + protected Class getEndpointClass() { + return ServiceActivatorEndpoint.class; + } + + @Override + protected Class getMethodInvokingAdapterClass() { + return MessageMappingMethodInvoker.class; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index 5b557823a2..3bcef2634b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -75,4 +75,8 @@ public abstract class AbstractDispatcher implements MessageDispatcher { return this.messageExchangeTemplate.send(message, target); } + public String toString() { + return this.getClass().getSimpleName() + " with targets: " + this.targets; + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java new file mode 100644 index 0000000000..d38d4bfd30 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java @@ -0,0 +1,181 @@ +/* + * Copyright 2002-2008 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.endpoint; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.CompositeMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHeaders; +import org.springframework.integration.message.MessageRejectedException; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.MessagingException; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * @author Mark Fisher + */ +public abstract class AbstractInOutEndpoint extends AbstractEndpoint { + + private volatile MessageSelector selector; + + private final List interceptors = new CopyOnWriteArrayList(); + + + public void setSelector(MessageSelector selector) { + this.selector = selector; + } + + public void addInterceptor(EndpointInterceptor interceptor) { + this.interceptors.add(interceptor); + } + + public void setInterceptors(List interceptors) { + this.interceptors.clear(); + for (EndpointInterceptor interceptor : interceptors) { + this.addInterceptor(interceptor); + } + } + + @Override + protected boolean sendInternal(Message message) { + for (EndpointInterceptor interceptor : this.interceptors) { + message = interceptor.preHandle(message); + if (message == null) { + return false; + } + } + if (!this.supports(message)) { + throw new MessageRejectedException(message, "unsupported message"); + } + Object result = this.handle(message); + if (result == null) { + return false; + } + Message reply = buildReplyMessage(result, message.getHeaders()); + MessageTarget replyTarget = this.resolveReplyTarget(message); + if (reply instanceof CompositeMessage && this.shouldSplitComposite()) { + for (Message nextReply : (CompositeMessage) reply) { + this.sendReplyMessage(nextReply, replyTarget); + } + return true; + } + else { + return this.sendReplyMessage(reply, replyTarget); + } + } + + protected abstract Object handle(Message message); + + protected boolean supports(Message message) { + if (this.selector != null && !this.selector.accept(message)) { + if (logger.isDebugEnabled()) { + logger.debug("selector for endpoint '" + this + "' rejected message: " + message); + } + return false; + } + return true; + } + + protected boolean shouldSplitComposite() { + return false; + } + + private boolean sendReplyMessage(Message replyMessage, MessageTarget replyTarget) { + for (int i = this.interceptors.size() - 1; i >= 0; i--) { + EndpointInterceptor interceptor = this.interceptors.get(i); + if (interceptor != null) { + replyMessage = interceptor.postHandle(replyMessage); + if (replyMessage == null) { + return false; + } + } + } + return this.getMessageExchangeTemplate().send(replyMessage, replyTarget); + } + + private Message buildReplyMessage(Object result, MessageHeaders requestHeaders) { + MessageBuilder builder = null; + if (result instanceof MessageBuilder) { + builder = (MessageBuilder) result; + } + else if (result instanceof CompositeMessage) { + List> messages = ((CompositeMessage) result).getPayload(); + List> replies = new ArrayList>(); + for (Message message : messages) { + replies.add(this.buildReplyMessage(message, requestHeaders)); + } + return new CompositeMessage(replies); + } + else if (result instanceof Message) { + builder = MessageBuilder.fromMessage((Message) result); + } + else { + builder = MessageBuilder.fromPayload(result); + } + return builder.copyHeadersIfAbsent(requestHeaders) + .setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestHeaders.getId()) + .build(); + } + + private MessageTarget resolveReplyTarget(Message requestMessage) { + MessageTarget replyTarget = this.getTarget(); + if (replyTarget == null) { + Object returnAddress = requestMessage.getHeaders().getReturnAddress(); + if (returnAddress != null) { + if (returnAddress instanceof MessageTarget) { + replyTarget = (MessageTarget) returnAddress; + } + else if (returnAddress instanceof String) { + ChannelRegistry channelRegistry = this.getChannelRegistry(); + if (channelRegistry != null) { + replyTarget = channelRegistry.lookupChannel((String) returnAddress); + } + } + } + } + if (replyTarget == null) { + throw new MessagingException("unable to resolve reply target"); + } + return replyTarget; + } + + // TODO: remove these methods after refactoring + + private volatile String inputChannelName; + + public String getInputChannelName() { + return this.inputChannelName; + } + + public void setInputChannelName(String inputChannelName) { + this.inputChannelName = inputChannelName; + } + + public String getOutputChannelName() { + if (this.getTarget() instanceof MessageChannel) { + return ((MessageChannel) this.getTarget()).getName(); + } + return null; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java new file mode 100644 index 0000000000..defb04d354 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java @@ -0,0 +1,62 @@ +/* + * Copyright 2002-2008 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.endpoint; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMappingMethodInvoker; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + */ +public class ServiceActivatorEndpoint extends AbstractInOutEndpoint implements InitializingBean { + + private final MessageMappingMethodInvoker invoker; + + private final MessageHandler handler; + + + public ServiceActivatorEndpoint(MessageMappingMethodInvoker invoker) { + Assert.notNull(invoker, "invoker must not be null"); + this.invoker = invoker; + this.handler = null; + } + + public ServiceActivatorEndpoint(MessageHandler handler) { + Assert.notNull(handler, "handler must not be null"); + this.handler = handler; + this.invoker = null; + } + + + public void afterPropertiesSet() throws Exception { + if (this.invoker != null) { + this.invoker.afterPropertiesSet(); + } + } + + @Override + protected Object handle(Message message) { + if (this.invoker != null) { + return this.invoker.invokeMethod(message); + } + return this.handler.handle(message); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/SplitterEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/SplitterEndpoint.java index c6e3fe4757..c746ec3f4a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/SplitterEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/SplitterEndpoint.java @@ -18,18 +18,15 @@ package org.springframework.integration.splitter; import java.util.List; -import org.springframework.integration.channel.ChannelRegistry; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.AbstractInOutEndpoint; +import org.springframework.integration.message.CompositeMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.MessagingException; import org.springframework.util.Assert; /** * @author Mark Fisher */ -public class SplitterEndpoint extends AbstractEndpoint { +public class SplitterEndpoint extends AbstractInOutEndpoint { private final Splitter splitter; @@ -40,57 +37,18 @@ public class SplitterEndpoint extends AbstractEndpoint { } - // TODO: move to superclass - private MessageTarget resolveReplyTarget(Object returnAddress) { - MessageTarget replyTarget = this.getTarget(); - if (replyTarget == null && returnAddress != null) { - if (returnAddress instanceof MessageTarget) { - replyTarget = (MessageTarget) returnAddress; - } - else if (returnAddress instanceof String) { - ChannelRegistry channelRegistry = this.getChannelRegistry(); - if (channelRegistry != null) { - replyTarget = channelRegistry.lookupChannel((String) returnAddress); - } - } - } - if (replyTarget == null) { - throw new MessagingException("unable to resolve reply target"); - } - return replyTarget; + @Override + protected boolean shouldSplitComposite() { + return true; } @Override - protected boolean sendInternal(Message message) { + protected Message handle(Message message) { List> results = this.splitter.split(message); - if (results != null) { - for (Message splitMessage : results) { - MessageTarget replyTarget = this.resolveReplyTarget(message.getHeaders().getReturnAddress()); - this.getMessageExchangeTemplate().send(splitMessage, replyTarget); - } - return true; + if (results == null || results.isEmpty()) { + return null; } - return false; - } - - - // TODO: remove these methods after refactoring - - private volatile String inputChannelName; - - public String getInputChannelName() { - return this.inputChannelName; - } - - public void setInputChannelName(String inputChannelName) { - this.inputChannelName = inputChannelName; - } - - public String getOutputChannelName() { - if (this.getTarget() instanceof MessageChannel) { - return ((MessageChannel) this.getTarget()).getName(); - } - return null; + return new CompositeMessage(results); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java index 2855155dd1..0038190cbb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java @@ -47,7 +47,7 @@ public class TestHandler implements MessageHandler { this.replyMessageText = replyMessageText; } - public Message handle(Message message) { + public Message handle(Message message) { this.messageString = message.getPayload().toString(); this.latch.countDown(); return (this.replyMessageText != null) ? new StringMessage(this.replyMessageText) : null; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml index 2e996c81fe..f3c32e6e78 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml @@ -17,7 +17,7 @@ input-channel="testChannel" ref="testHandler" output-channel="replyChannel"> - + @@ -28,7 +28,7 @@ input-channel="testChannel" ref="testHandler" output-channel="replyChannel"> - + @@ -36,7 +36,9 @@ - + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java similarity index 51% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java index a0692be40a..cb63fbd993 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.endpoint.interceptor; +package org.springframework.integration.dispatcher; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -25,25 +25,20 @@ import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; import org.springframework.transaction.IllegalTransactionStateException; -import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.annotation.Propagation; -import org.springframework.transaction.support.TransactionCallback; -import org.springframework.transaction.support.TransactionTemplate; /** * @author Mark Fisher */ -public class TransactionInterceptorTests { +public class PollingTransactionTests { @Test - public void testTransactionInterceptorWithCommit() throws InterruptedException { + public void transactionWithCommit() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "transactionInterceptorTests.xml", this.getClass()); + "transactionTests.xml", this.getClass()); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); MessageChannel input = (MessageChannel) context.getBean("goodInput"); PollableChannel output = (PollableChannel) context.getBean("output"); @@ -51,16 +46,16 @@ public class TransactionInterceptorTests { assertEquals(0, txManager.getRollbackCount()); input.send(new StringMessage("test")); txManager.waitForCompletion(1000); - Message message = output.receive(500); + Message message = output.receive(0); assertNotNull(message); assertEquals(1, txManager.getCommitCount()); assertEquals(0, txManager.getRollbackCount()); } @Test - public void testTransactionInterceptorWithRollback() throws InterruptedException { + public void transactionWithRollback() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "transactionInterceptorTests.xml", this.getClass()); + "transactionTests.xml", this.getClass()); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); MessageChannel input = (MessageChannel) context.getBean("badInput"); PollableChannel output = (PollableChannel) context.getBean("output"); @@ -68,105 +63,90 @@ public class TransactionInterceptorTests { assertEquals(0, txManager.getRollbackCount()); input.send(new StringMessage("test")); txManager.waitForCompletion(1000); - Message message = output.receive(500); + Message message = output.receive(0); assertNull(message); assertEquals(0, txManager.getCommitCount()); assertEquals(1, txManager.getRollbackCount()); } @Test - public void testPropagationRequired() { + public void propagationRequired() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "transactionInterceptorPropagationTests.xml", this.getClass()); + "propagationRequiredTests.xml", this.getClass()); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); - final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("required"); + PollableChannel input = (PollableChannel) context.getBean("input"); + PollableChannel output = (PollableChannel) context.getBean("output"); assertEquals(0, txManager.getCommitCount()); - endpoint.send(new StringMessage("test")); + input.send(new StringMessage("test")); + Message reply = output.receive(3000); + assertNotNull(reply); + txManager.waitForCompletion(3000); assertEquals(1, txManager.getCommitCount()); - TestTransactionManager outerTxManager = new TestTransactionManager(); - TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - return endpoint.send(new StringMessage("test")); - } - }); - assertEquals(1, outerTxManager.getCommitCount()); - assertEquals(2, txManager.getCommitCount()); assertEquals(Propagation.REQUIRED.value(), txManager.getLastDefinition().getPropagationBehavior()); } @Test - public void testPropagationRequiresNew() { + public void propagationRequiresNew() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "transactionInterceptorPropagationTests.xml", this.getClass()); + "propagationRequiresNewTests.xml", this.getClass()); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); - final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("requiresNew"); + PollableChannel input = (PollableChannel) context.getBean("input"); + PollableChannel output = (PollableChannel) context.getBean("output"); assertEquals(0, txManager.getCommitCount()); - endpoint.send(new StringMessage("test")); + input.send(new StringMessage("test")); + Message reply = output.receive(3000); + assertNotNull(reply); + txManager.waitForCompletion(3000); assertEquals(1, txManager.getCommitCount()); - TestTransactionManager outerTxManager = new TestTransactionManager(); - TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - return endpoint.send(new StringMessage("test")); - } - }); - assertEquals(1, outerTxManager.getCommitCount()); - assertEquals(2, txManager.getCommitCount()); assertEquals(Propagation.REQUIRES_NEW.value(), txManager.getLastDefinition().getPropagationBehavior()); } @Test - public void testPropagationSupports() { + public void propagationSupports() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "transactionInterceptorPropagationTests.xml", this.getClass()); + "propagationSupportsTests.xml", this.getClass()); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); - final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("supports"); + PollableChannel input = (PollableChannel) context.getBean("input"); + PollableChannel output = (PollableChannel) context.getBean("output"); assertEquals(0, txManager.getCommitCount()); - endpoint.send(new StringMessage("test")); + input.send(new StringMessage("test")); + Message reply = output.receive(3000); + assertNotNull(reply); assertEquals(0, txManager.getCommitCount()); - TestTransactionManager outerTxManager = new TestTransactionManager(); - TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - return endpoint.send(new StringMessage("test")); - } - }); - assertEquals(0, txManager.getCommitCount()); - assertEquals(1, outerTxManager.getCommitCount()); + assertNull(txManager.getLastDefinition()); } @Test - public void testPropagationNotSupported() { + public void propagationNotSupported() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "transactionInterceptorPropagationTests.xml", this.getClass()); + "propagationNotSupportedTests.xml", this.getClass()); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); - final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("notSupported"); + PollableChannel input = (PollableChannel) context.getBean("input"); + PollableChannel output = (PollableChannel) context.getBean("output"); assertEquals(0, txManager.getCommitCount()); - endpoint.send(new StringMessage("test")); + input.send(new StringMessage("test")); + Message reply = output.receive(3000); + assertNotNull(reply); assertEquals(0, txManager.getCommitCount()); - TestTransactionManager outerTxManager = new TestTransactionManager(); - TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - return endpoint.send(new StringMessage("test")); - } - }); - assertEquals(0, txManager.getCommitCount()); - assertEquals(1, outerTxManager.getCommitCount()); + assertNull(txManager.getLastDefinition()); } - @Test(expected = IllegalTransactionStateException.class) - public void testPropagationMandatoryCalledWithoutTransaction() throws Throwable { + @Test + public void propagationMandatory() throws Throwable { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "transactionInterceptorPropagationTests.xml", this.getClass()); - final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("mandatory"); - try { - endpoint.send(new StringMessage("test")); - } - catch (MessageHandlingException e) { - throw e.getCause(); - } + "propagationMandatoryTests.xml", this.getClass()); + TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); + PollableChannel input = (PollableChannel) context.getBean("input"); + PollableChannel output = (PollableChannel) context.getBean("output"); + PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel"); + assertEquals(0, txManager.getCommitCount()); + input.send(new StringMessage("test")); + Message errorMessage = errorChannel.receive(3000); + assertNotNull(errorMessage); + Object payload = errorMessage.getPayload(); + assertEquals(IllegalTransactionStateException.class, payload.getClass()); + assertNull(output.receive(0)); + assertEquals(0, txManager.getCommitCount()); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestBean.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/TestBean.java similarity index 92% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestBean.java rename to org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/TestBean.java index 25b702a5f6..6c354844c6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestBean.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/TestBean.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.endpoint.interceptor; +package org.springframework.integration.dispatcher; /** * @author Mark Fisher diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestTransactionManager.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/TestTransactionManager.java similarity index 97% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestTransactionManager.java rename to org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/TestTransactionManager.java index dcae0b1269..0972060a26 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestTransactionManager.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/TestTransactionManager.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.endpoint.interceptor; +package org.springframework.integration.dispatcher; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationMandatoryTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationMandatoryTests.xml new file mode 100644 index 0000000000..893d15b4d9 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationMandatoryTests.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationNotSupportedTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationNotSupportedTests.xml new file mode 100644 index 0000000000..3e91b99348 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationNotSupportedTests.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiredTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiredTests.xml new file mode 100644 index 0000000000..d8ccca3ab8 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiredTests.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiresNewTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiresNewTests.xml new file mode 100644 index 0000000000..fab451b8f9 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiresNewTests.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationSupportsTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationSupportsTests.xml new file mode 100644 index 0000000000..89c913e728 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationSupportsTests.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml similarity index 74% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorTests.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml index 58b2b6716c..b8cede20d0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml @@ -17,24 +17,22 @@ ref="testBean" method="bad" output-channel="output"> - - - - + + + - - - - + + + - + - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorPropagationTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorPropagationTests.xml deleted file mode 100644 index 221b577eec..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorPropagationTests.xml +++ /dev/null @@ -1,74 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -