Added AbstractInOutEndpoint and ServiceActivatorEndpoint. SplitterEndpoint now extends AbstractInOutEndpoint as well.

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) {

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<? extends MessageHandler> getHandlerAdapterClass() {
-		return DefaultMessageHandler.class;
+	protected Class<? extends MessageEndpoint> getEndpointClass() {
+		return ServiceActivatorEndpoint.class;
+	}
+
+	@Override
+	protected Class<?> getMethodInvokingAdapterClass() {
+		return MessageMappingMethodInvoker.class;
 	}
 
 }

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;
+	}
+
 }

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<EndpointInterceptor> interceptors = new CopyOnWriteArrayList<EndpointInterceptor>();
+
+
+	public void setSelector(MessageSelector selector) {
+		this.selector = selector;
+	}
+
+	public void addInterceptor(EndpointInterceptor interceptor) {
+		this.interceptors.add(interceptor);
+	}
+
+	public void setInterceptors(List<EndpointInterceptor> 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<Message<?>> messages = ((CompositeMessage) result).getPayload();
+			List<Message<?>> replies = new ArrayList<Message<?>>();
+			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;
+	}
+
+}

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);
+	}
+
+}

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<Message<?>> 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);
 	}
 
 }

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;

org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml

@@ -17,7 +17,7 @@
 	                   input-channel="testChannel"
 	                   ref="testHandler"
 	                   output-channel="replyChannel">
-		<poller period="100"/>
+		<poller period="10000" max-messages-per-poll="1"/>
 		<interceptors>
 			<beans:bean class="org.springframework.integration.config.TestPreHandleInterceptor"/>
 			<beans:bean class="org.springframework.integration.config.TestPostHandleInterceptor"/>
@@ -28,7 +28,7 @@
 	                   input-channel="testChannel"
 	                   ref="testHandler"
 	                   output-channel="replyChannel">
-		<poller period="100"/>
+		<poller period="10000" max-messages-per-poll="1"/>
 		<interceptors>
 			<ref bean="preInterceptor"/>
 			<ref bean="postInterceptor"/>
@@ -36,7 +36,9 @@
 	</service-activator>
 
 
-	<beans:bean id="testHandler" class="org.springframework.integration.config.TestHandler"/>
+	<beans:bean id="testHandler" class="org.springframework.integration.config.TestHandler">
+		<beans:property name="replyMessageText" value="test"/>
+	</beans:bean>
 
 	<beans:bean id="preInterceptor" class="org.springframework.integration.config.TestPreHandleInterceptor"/>
 

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());
 	}
 
 }

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

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;

org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationMandatoryTests.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="input"/>
+	<channel id="output"/>
+	<channel id="errorChannel"/>
+
+	<service-activator id="mandatory"
+	                   input-channel="input"
+	                   ref="testBean"
+	                   method="good"
+                       output-channel="output">
+        <poller period="10000" max-messages-per-poll="1">
+        	<transactional transaction-manager="txManager" propagation="MANDATORY"/>
+        </poller>
+    </service-activator>
+
+	<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
+
+	<beans:bean id="txManager" class="org.springframework.integration.dispatcher.TestTransactionManager"/>
+
+</beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationNotSupportedTests.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="input"/>
+	<channel id="output"/>
+
+	<service-activator id="notSupported"
+	                   input-channel="input"
+	                   ref="testBean"
+	                   method="good"
+                       output-channel="output">
+        <poller period="10000" max-messages-per-poll="1">
+        	<transactional transaction-manager="txManager" propagation="NOT_SUPPORTED"/>
+        </poller>
+    </service-activator>
+
+	<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
+
+	<beans:bean id="txManager" class="org.springframework.integration.dispatcher.TestTransactionManager"/>
+
+</beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiredTests.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="input"/>
+	<channel id="output"/>
+
+	<service-activator id="required"
+	                   input-channel="input"
+	                   ref="testBean"
+	                   method="good"
+                       output-channel="output">
+        <poller period="10000" max-messages-per-poll="1">
+        	<transactional transaction-manager="txManager" propagation="REQUIRED"/>
+        </poller>
+    </service-activator>
+
+	<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
+
+	<beans:bean id="txManager" class="org.springframework.integration.dispatcher.TestTransactionManager"/>
+
+</beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationRequiresNewTests.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="input"/>
+	<channel id="output"/>
+
+	<service-activator id="requiresNew"
+	                   input-channel="input"
+	                   ref="testBean"
+	                   method="good"
+                       output-channel="output">
+        <poller period="10000" max-messages-per-poll="1">
+        	<transactional transaction-manager="txManager" propagation="REQUIRES_NEW"/>
+        </poller>
+    </service-activator>
+
+	<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
+
+	<beans:bean id="txManager" class="org.springframework.integration.dispatcher.TestTransactionManager"/>
+
+</beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/propagationSupportsTests.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="input"/>
+	<channel id="output"/>
+
+	<service-activator id="supports"
+	                   input-channel="input"
+	                   ref="testBean"
+	                   method="good"
+                       output-channel="output">
+        <poller period="10000" max-messages-per-poll="1">
+        	<transactional transaction-manager="txManager" propagation="SUPPORTS"/>
+        </poller>
+    </service-activator>
+
+	<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
+
+	<beans:bean id="txManager" class="org.springframework.integration.dispatcher.TestTransactionManager"/>
+
+</beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml

@@ -17,24 +17,22 @@
 	                  ref="testBean"
 	                  method="bad"
                       output-channel="output">
-        <poller period="100"/>
-    	<interceptors>
-			<transaction-interceptor transaction-manager="txManager"/>
-    	</interceptors>
+        <poller period="10000" max-messages-per-poll="1">
+        	<transactional transaction-manager="txManager"/>
+        </poller>
     </service-activator>
 
 	<service-activator input-channel="goodInput"
 	                  ref="testBean"
 	                  method="good"
                       output-channel="output">
-        <poller period="100"/>
-		<interceptors>
-			<transaction-interceptor transaction-manager="txManager" propagation="REQUIRED"/>
-		</interceptors>
+        <poller period="10000" max-messages-per-poll="1">
+        	<transactional transaction-manager="txManager" propagation="REQUIRED"/>
+        </poller>
 	</service-activator>
 
-	<beans:bean id="testBean" class="org.springframework.integration.endpoint.interceptor.TestBean"/>
+	<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
 
-	<beans:bean id="txManager" class="org.springframework.integration.endpoint.interceptor.TestTransactionManager"/>
+	<beans:bean id="txManager" class="org.springframework.integration.dispatcher.TestTransactionManager"/>
 
 </beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorPropagationTests.xml

@@ -1,74 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans:beans xmlns="http://www.springframework.org/schema/integration"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:beans="http://www.springframework.org/schema/beans"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans
-			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
-			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
-
-	<message-bus/>
-
-	<channel id="input"/>
-	<channel id="output"/>
-
-	<service-activator id="required"
-	                  input-channel="input"
-	                  ref="testBean"
-	                  method="good"
-                      output-channel="output">
-        <poller period="10000"/>
-    	<interceptors>
-			<transaction-interceptor transaction-manager="txManager" propagation="REQUIRED"/>
-    	</interceptors>
-    </service-activator>
-
-	<service-activator id="requiresNew"
-	                  input-channel="input"
-	                  ref="testBean"
-	                  method="good"
-                      output-channel="output">
-        <poller period="10000"/>
-    	<interceptors>
-			<transaction-interceptor transaction-manager="txManager" propagation="REQUIRES_NEW"/>
-    	</interceptors>
-    </service-activator>
-
-	<service-activator id="supports"
-	                  input-channel="input"
-	                  ref="testBean"
-	                  method="good"
-                      output-channel="output">
-        <poller period="10000"/>
-    	<interceptors>
-			<transaction-interceptor transaction-manager="txManager" propagation="SUPPORTS"/>
-    	</interceptors>
-    </service-activator>
-
-	<service-activator id="notSupported"
-	                  input-channel="input"
-	                  ref="testBean"
-	                  method="good"
-                      output-channel="output">
-        <poller period="10000"/>
-    	<interceptors>
-			<transaction-interceptor transaction-manager="txManager" propagation="NOT_SUPPORTED"/>
-    	</interceptors>
-    </service-activator>
-
-	<service-activator id="mandatory"
-	                  input-channel="input"
-	                  ref="testBean"
-	                  method="good"
-                      output-channel="output">
-        <poller period="10000"/>
-    	<interceptors>
-			<transaction-interceptor transaction-manager="txManager" propagation="MANDATORY"/>
-    	</interceptors>
-    </service-activator>
-
-	<beans:bean id="testBean" class="org.springframework.integration.endpoint.interceptor.TestBean"/>
-
-	<beans:bean id="txManager" class="org.springframework.integration.endpoint.interceptor.TestTransactionManager"/>
-
-</beans:beans>