INT-3114 Call MessageHandlers Directly

Invoke ARPMH Directly (SA, Splitter etc)

If a bean referenced by a service activator etc, is already an ARPMH,
then wire it directly into the endpoint rather than wrapping it in a
MethodInvoking Handler.

Disallow the same bean being used in this way in multiple endpoints.
It would cause rewiring of outputChannel etc.
Detect such use and suggest using prototype scope.

Previously, an ARPMH used in a MethodInvokingHandler didn't work
in some cases, because the "inner" ARPMH had no output-channel
and the outer ARPMH (e.g. transformer) wouldn't get a reply and
would complain if requiresReply is set.

The general solution is to detect the ARPMH targetObject in
AbstractStandardMessageHandlerFactoryBean, invoke canBeUsedDirect()
to allow the subclass to vote - specifically it will be allowed
if the type is not also a special type understood by the subclass
and no endpoint-specific attributes are set.

To this end, remove a few default attributes from the schema to
enable detection of explicit setting of these attributes.

If these conditions pass and the method is handleRequest (either
explicitly or by omission) then wire it in and invoke the
subclass to post process the handler (set attributes).

Restructure this post processing in each subclass so that it can be invoked in
this way, or directly by the subclass when it makes a custom
handler.

Polishing - PR Comments

Mainly a little refactoring in the RouterFactoryBean to allow
the parameter in canBeUsedDirect() to be an ARPMH instead of
Object.

Plus a couple more minor changes.

spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java

@@ -16,6 +16,8 @@ package org.springframework.integration.config;
 import java.util.List;
 
 import org.aopalliance.aop.Advice;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
@@ -41,6 +43,8 @@ import org.springframework.util.CollectionUtils;
 public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageHandler>
 		implements FactoryBean<MessageHandler>, BeanFactoryAware {
 
+	protected final Log logger = LogFactory.getLog(this.getClass());
+
 	private volatile H handler;
 
 	private volatile MessageChannel outputChannel;

spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2002-2010 the original author or authors.
- * 
+ * Copyright 2002-2013 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.
@@ -13,6 +13,9 @@
 
 package org.springframework.integration.config;
 
+import java.util.HashSet;
+import java.util.Set;
+
 import org.springframework.aop.TargetSource;
 import org.springframework.aop.framework.Advised;
 import org.springframework.expression.Expression;
@@ -20,27 +23,33 @@ import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.SpelParserConfiguration;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
  * Base class for FactoryBeans that create MessageHandler instances.
- * 
+ *
  * @author Mark Fisher
  * @author Alexander Peters
+ * @author Gary Russell
  */
-abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean<MessageHandler>{
+abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean<MessageHandler> {
 
 	private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true,
 			true));
 
+	private static final Set<MessageHandler> referencedReplyProducers = new HashSet<MessageHandler>();
+
 	private volatile Object targetObject;
 
 	private volatile String targetMethodName;
 
 	private volatile Expression expression;
 
+	private volatile String beanName;
+
 	public void setTargetObject(Object targetObject) {
 		this.targetObject = targetObject;
 	}
@@ -57,6 +66,7 @@ abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleM
 		this.expression = expression;
 	}
 
+	@Override
 	protected MessageHandler createHandler() {
 		MessageHandler handler;
 		if (this.targetObject == null) {
@@ -66,19 +76,55 @@ abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleM
 		if (this.targetObject != null) {
 			Assert.state(this.expression == null,
 					"The 'targetObject' and 'expression' properties are mutually exclusive.");
+			boolean targetIsDirectReplyProducingHandler = this.extractTypeIfPossible(targetObject,
+										AbstractReplyProducingMessageHandler.class) != null
+							&& this.canBeUsedDirect(
+									(AbstractReplyProducingMessageHandler) targetObject) // give subclasses a say
+							&& this.methodIsHandleMessageOrEmpty(this.targetMethodName);
 			if (this.targetObject instanceof MessageProcessor<?>) {
 				handler = this.createMessageProcessingHandler((MessageProcessor<?>) this.targetObject);
-			} else {
+			}
+			else if (targetIsDirectReplyProducingHandler) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Wiring handler (" + beanName + ") directly into endpoint");
+				}
+				handler = (MessageHandler) targetObject;
+				this.checkReuse((AbstractReplyProducingMessageHandler) handler);
+				this.postProcessReplyProducer((AbstractReplyProducingMessageHandler) handler);
+			}
+			else {
 				handler = this.createMethodInvokingHandler(this.targetObject, this.targetMethodName);
 			}
-		} else if (this.expression != null) {
+		}
+		else if (this.expression != null) {
 			handler = this.createExpressionEvaluatingHandler(this.expression);
-		} else {
+		}
+		else {
 			handler = this.createDefaultHandler();
 		}
 		return handler;
 	}
 
+	protected void checkForIllegalTarget(Object targetObject, String targetMethodName) {
+		if (targetObject instanceof AbstractReplyProducingMessageHandler
+				&& this.methodIsHandleMessageOrEmpty(targetMethodName)) {
+			/*
+			 * If we allow an ARPMH to be the target of another ARPMH, the reply would
+			 * be attempted to be sent by the inner (no output channel) and a reply would
+			 * never be received by the outer (fails if replyRequired).
+			 */
+			throw new IllegalArgumentException("AbstractReplyProducingMessageHandler.handleMessage() "
+					+ "is not allowed for a MethodInvokingHandler");
+		}
+	}
+
+	private void checkReuse(AbstractReplyProducingMessageHandler replyHandler) {
+		Assert.isTrue(!referencedReplyProducers.contains(targetObject),
+				"An AbstractReplyProducingMessageHandler may only be referenced once (" +
+				replyHandler.getComponentName() + ") - use scope=\"prototype\"");
+		referencedReplyProducers.add(replyHandler);
+	}
+
 	/**
 	 * Subclasses must implement this method to create the MessageHandler.
 	 */
@@ -118,4 +164,16 @@ abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleM
 		return null;
 	}
 
+	protected boolean methodIsHandleMessageOrEmpty(String targetMethodName) {
+		return (!StringUtils.hasText(targetMethodName)
+				|| "handleMessage".equals(targetMethodName));
+	}
+
+	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+		return false;
+	}
+
+	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java

@@ -23,6 +23,8 @@ import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
 import org.springframework.integration.filter.MessageFilter;
 import org.springframework.integration.filter.MethodInvokingSelector;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
@@ -65,6 +67,7 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 			selector = (MessageSelector) targetObject;
 		}
 		else if (StringUtils.hasText(targetMethodName)) {
+			this.checkForIllegalTarget(targetObject, targetMethodName);
 			selector = new MethodInvokingSelector(targetObject, targetMethodName);
 		}
 		else {
@@ -73,6 +76,15 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 		return this.createFilter(selector);
 	}
 
+	@Override
+	protected void checkForIllegalTarget(Object targetObject, String targetMethodName) {
+		if (targetObject instanceof AbstractReplyProducingMessageHandler
+				&& this.methodIsHandleMessageOrEmpty(targetMethodName)) {
+			throw new IllegalArgumentException("You cannot use 'AbstractReplyProducingMessageHandler.handleMessage()' "
+					+ "as a filter - it does not return a result");
+		}
+	}
+
 	@Override
 	MessageHandler createExpressionEvaluatingHandler(Expression expression) {
 		return this.createFilter(new ExpressionEvaluatingSelector(expression));
@@ -80,19 +92,52 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 
 	private MessageFilter createFilter(MessageSelector selector) {
 		MessageFilter filter = new MessageFilter(selector);
+		postProcessReplyProducer(filter);
+		return filter;
+	}
+
+	private void postProcessFilter(MessageFilter filter) {
 		if (this.throwExceptionOnRejection != null) {
 			filter.setThrowExceptionOnRejection(this.throwExceptionOnRejection);
 		}
 		if (this.discardChannel != null) {
 			filter.setDiscardChannel(discardChannel);
 		}
-		if (this.sendTimeout != null) {
-			filter.setSendTimeout(this.sendTimeout.longValue());
-		}
 		if (this.discardWithinAdvice != null) {
 			filter.setDiscardWithinAdvice(this.discardWithinAdvice);
 		}
-		return filter;
 	}
 
+
+	@Override
+	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+		if (this.sendTimeout != null) {
+			handler.setSendTimeout(this.sendTimeout.longValue());
+		}
+		if (!(handler instanceof MessageFilter)) {
+			Assert.isNull(this.throwExceptionOnRejection, "Cannot set throwExceptionOnRejection if the referenced bean is "
+					+ "an AbstractReplyProducingMessageHandler, but not a MessageFilter");
+			Assert.isNull(this.discardChannel, "Cannot set discardChannel if the referenced bean is "
+					+ "an AbstractReplyProducingMessageHandler, but not a MessageFilter");
+			Assert.isNull(this.discardWithinAdvice, "Cannot set discardWithinAdvice if the referenced bean is "
+					+ "an AbstractReplyProducingMessageHandler, but not a MessageFilter");
+		}
+		else {
+			postProcessFilter((MessageFilter) handler);
+		}
+	}
+
+	/**
+	 * MessageFilter is an ARPMH. If a non-MessageFilter ARPMH is also a
+	 * MessageSelector, MesageSelector wins and gets wrapped in a MessageFilter.
+	 */
+	@Override
+	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+		return handler instanceof MessageFilter
+				|| (!(handler instanceof MessageSelector)
+						&& this.discardChannel == null && this.throwExceptionOnRejection == null
+						&& this.discardWithinAdvice == null);
+	}
+
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2002-2011 the original author or authors.
- * 
+ * Copyright 2002-2013 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.
@@ -21,6 +21,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.expression.Expression;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.router.AbstractMappingMessageRouter;
 import org.springframework.integration.router.AbstractMessageRouter;
 import org.springframework.integration.router.ExpressionEvaluatingRouter;
@@ -31,14 +32,15 @@ import org.springframework.util.StringUtils;
 
 /**
  * Factory bean for creating a Message Router.
- * 
+ *
  * @author Mark Fisher
  * @author Jonas Partner
  * @author Oleg Zhurakousky
  * @author Dave Syer
+ * @author Gary Russell
  */
 public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
-	
+
 	private final Log logger = LogFactory.getLog(this.getClass());
 
 	private volatile Map<String, String> channelMappings;
@@ -52,7 +54,7 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 	private volatile Boolean applySequence;
 
 	private volatile Boolean ignoreSendFailures;
-	
+
 	private volatile ChannelResolver channelResolver;
 
 
@@ -79,7 +81,7 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 	public void setIgnoreSendFailures(Boolean ignoreSendFailures) {
 		this.ignoreSendFailures = ignoreSendFailures;
 	}
-	
+
 	public void setChannelMappings(Map<String, String> channelMappings) {
 		this.channelMappings = channelMappings;
 	}
@@ -89,6 +91,10 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 		Assert.notNull(targetObject, "target object must not be null");
 		AbstractMessageRouter router = this.extractTypeIfPossible(targetObject, AbstractMessageRouter.class);
 		if (router == null) {
+			if (targetObject instanceof MessageHandler && this.noRouterAttributesProvided()
+					&& this.methodIsHandleMessageOrEmpty(targetMethodName)) {
+				return (MessageHandler) targetObject;
+			}
 			router = this.createMethodInvokingRouter(targetObject, targetMethodName);
 			this.configureRouter(router);
 		}
@@ -109,7 +115,7 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 	}
 
 	private AbstractMappingMessageRouter createMethodInvokingRouter(Object targetObject, String targetMethodName) {
-		MethodInvokingRouter router = (StringUtils.hasText(targetMethodName)) 
+		MethodInvokingRouter router = (StringUtils.hasText(targetMethodName))
 				? new MethodInvokingRouter(targetObject, targetMethodName)
 				: new MethodInvokingRouter(targetObject);
 		return router;
@@ -147,4 +153,15 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 		}
 	}
 
+	@Override
+	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+		return noRouterAttributesProvided();
+	}
+
+	private boolean noRouterAttributesProvided() {
+		return this.channelMappings == null && this.defaultOutputChannel == null
+				&& this.timeout == null && this.resolutionRequired == null && this.applySequence == null
+				&& this.ignoreSendFailures == null && this.channelResolver == null;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,7 +17,9 @@
 package org.springframework.integration.config;
 
 import org.springframework.expression.Expression;
+import org.springframework.integration.Message;
 import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.integration.handler.ServiceActivatingHandler;
@@ -25,8 +27,9 @@ import org.springframework.util.StringUtils;
 
 /**
  * FactoryBean for creating {@link ServiceActivatingHandler} instances.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
 public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
@@ -45,10 +48,45 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 
 	@Override
 	MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) {
-		ServiceActivatingHandler handler = (StringUtils.hasText(targetMethodName))
-				? new ServiceActivatingHandler(targetObject, targetMethodName)
-				: new ServiceActivatingHandler(targetObject);
-		return this.configureHandler(handler);
+		MessageHandler handler = null;
+		handler = createDirectHandlerIfPossible(targetObject, targetMethodName);
+		if (handler == null) {
+			handler = configureHandler(
+					StringUtils.hasText(targetMethodName)
+					? new ServiceActivatingHandler(targetObject, targetMethodName)
+					: new ServiceActivatingHandler(targetObject));
+		}
+		return handler;
+	}
+
+	/**
+	 * If the target object is a {@link MessageHandler} and the method is 'handleMessage', return an
+	 * {@link AbstractReplyProducingMessageHandler} that wraps it.
+	 */
+	private MessageHandler createDirectHandlerIfPossible(final Object targetObject, String targetMethodName) {
+		MessageHandler handler = null;
+		if (targetObject instanceof MessageHandler
+				&& this.methodIsHandleMessageOrEmpty(targetMethodName)) {
+			if (targetObject instanceof AbstractReplyProducingMessageHandler) {
+				// should never happen but just return it if it's already an ARPMH
+				return (MessageHandler) targetObject;
+			}
+			/*
+			 * Return a reply-producing message handler so that we still get 'produced no reply' messages
+			 * and the super class will inject the advice chain to advise the handler if needed.
+			 */
+			handler = new AbstractReplyProducingMessageHandler() {
+
+				@Override
+				protected Object handleRequestMessage(Message<?> requestMessage) {
+
+					((MessageHandler) targetObject).handleMessage(requestMessage);
+					return null;
+				}
+			};
+
+		}
+		return handler;
 	}
 
 	@Override
@@ -63,14 +101,29 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 		return this.configureHandler(new ServiceActivatingHandler(processor));
 	}
 
-	private ServiceActivatingHandler configureHandler(ServiceActivatingHandler handler) {
+	private MessageHandler configureHandler(ServiceActivatingHandler handler) {
+		postProcessReplyProducer(handler);
+		return handler;
+	}
+
+
+	/**
+	 * Always returns true - any {@link AbstractReplyProducingMessageHandler} can
+	 * be used directly.
+	 */
+	@Override
+	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+		return true;
+	}
+
+	@Override
+	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
 		if (this.sendTimeout != null) {
 			handler.setSendTimeout(this.sendTimeout);
 		}
 		if (this.requiresReply != null) {
 			handler.setRequiresReply(this.requiresReply);
 		}
-		return handler;
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.config;
 
 import org.springframework.expression.Expression;
 import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.integration.splitter.DefaultMessageSplitter;
 import org.springframework.integration.splitter.ExpressionEvaluatingSplitter;
@@ -27,17 +28,18 @@ import org.springframework.util.StringUtils;
 
 /**
  * Factory bean for creating a Message Splitter.
- * 
+ *
  * @author Mark Fisher
  * @author Iwein Fuld
+ * @author Gary Russell
  */
 public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
 
 	private volatile Long sendTimeout;
 
-	private volatile boolean requiresReply;
+	private volatile Boolean requiresReply;
 
-	private volatile boolean applySequence = true;
+	private volatile Boolean applySequence;
 
 	private volatile String delimiters;
 
@@ -67,6 +69,7 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe
 		Assert.notNull(targetObject, "targetObject must not be null");
 		AbstractMessageSplitter splitter = this.extractTypeIfPossible(targetObject, AbstractMessageSplitter.class);
 		if (splitter == null) {
+			this.checkForIllegalTarget(targetObject, targetMethodName);
 			splitter = this.createMethodInvokingSplitter(targetObject, targetMethodName);
 			this.configureSplitter(splitter);
 		}
@@ -98,17 +101,42 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe
 	}
 
 	private AbstractMessageSplitter configureSplitter(AbstractMessageSplitter splitter) {
-		if (this.sendTimeout != null) {
-			splitter.setSendTimeout(sendTimeout);
-		}
-		if (this.delimiters != null) {
-			Assert.isTrue(splitter instanceof DefaultMessageSplitter, "The 'delimiters' property is only available" +
-					" for a Splitter definition where no 'ref', 'expression', or inner bean has been provided.");
-			((DefaultMessageSplitter) splitter).setDelimiters(this.delimiters);
-		}
-		splitter.setRequiresReply(requiresReply);
-		splitter.setApplySequence(applySequence);
+		this.postProcessReplyProducer(splitter);
 		return splitter;
 	}
 
+	@Override
+	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+		return handler instanceof AbstractMessageSplitter
+				|| (this.applySequence == null && this.delimiters == null);
+	}
+
+	@Override
+	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+		if (this.sendTimeout != null) {
+			handler.setSendTimeout(sendTimeout);
+		}
+		if (this.requiresReply != null) {
+			handler.setRequiresReply(requiresReply);
+		}
+		if (!(handler instanceof AbstractMessageSplitter)) {
+			Assert.isNull(this.applySequence, "Cannot set applySequence if the referenced bean is "
+					+ "an AbstractReplyProducingMessageHandler, but not an AbstractMessageSplitter");
+			Assert.isNull(this.delimiters, "Cannot set delimiters if the referenced bean is not an "
+					+ "an AbstractReplyProducingMessageHandler, but not an AbstractMessageSplitter");
+		}
+		else {
+			AbstractMessageSplitter splitter = (AbstractMessageSplitter) handler;
+			if (this.delimiters != null) {
+				Assert.isTrue(splitter instanceof DefaultMessageSplitter, "The 'delimiters' property is only available" +
+						" for a Splitter definition where no 'ref', 'expression', or inner bean has been provided.");
+				((DefaultMessageSplitter) splitter).setDelimiters(this.delimiters);
+			}
+			if (this.applySequence != null) {
+				splitter.setApplySequence(applySequence);
+			}
+		}
+	}
+
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2009 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.config;
 
 import org.springframework.expression.Expression;
 import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
 import org.springframework.integration.transformer.MessageTransformingHandler;
 import org.springframework.integration.transformer.MethodInvokingTransformer;
@@ -27,8 +28,9 @@ import org.springframework.util.StringUtils;
 
 /**
  * Factory bean for creating a Message Transformer.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
 
@@ -45,11 +47,14 @@ public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactor
 		if (targetObject instanceof Transformer) {
 			transformer = (Transformer) targetObject;
 		}
-		else if (StringUtils.hasText(targetMethodName)) {
-			transformer = new MethodInvokingTransformer(targetObject, targetMethodName);
-		}
 		else {
-			transformer = new MethodInvokingTransformer(targetObject);
+			this.checkForIllegalTarget(targetObject, targetMethodName);
+			if (StringUtils.hasText(targetMethodName)) {
+				transformer = new MethodInvokingTransformer(targetObject, targetMethodName);
+			}
+			else {
+				transformer = new MethodInvokingTransformer(targetObject);
+			}
 		}
 		return this.createHandler(transformer);
 	}
@@ -62,10 +67,25 @@ public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactor
 
 	private MessageTransformingHandler createHandler(Transformer transformer) {
 		MessageTransformingHandler handler = new MessageTransformingHandler(transformer);
-		if (this.sendTimeout != null) {
-			handler.setSendTimeout(this.sendTimeout.longValue());
-		}
+		this.postProcessReplyProducer(handler);
 		return handler;
 	}
 
+	@Override
+	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+		if (this.sendTimeout != null) {
+			handler.setSendTimeout(this.sendTimeout.longValue());
+		}
+	}
+
+	/**
+	 * Always returns true - any {@link AbstractReplyProducingMessageHandler} can
+	 * be used directly.
+	 */
+	@Override
+	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+		return true; // Any ARPMH can be a transformer
+	}
+
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractDelegatingConsumerEndpointParser.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.config.xml;
 
+import org.w3c.dom.Element;
+
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
@@ -23,7 +25,6 @@ import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.expression.DynamicExpression;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
-import org.w3c.dom.Element;
 
 /**
  * Base parser class for endpoints that delegate to a method invoker or

spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -25,6 +25,7 @@ import org.springframework.integration.annotation.ServiceActivator;
 /**
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Gary Russell
  */
 public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandler {
 
@@ -74,6 +75,7 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl
 		}
 	}
 
+	@Override
 	public String toString() {
 		return "ServiceActivator for [" + this.processor + "]";
 	}

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd

@@ -2480,7 +2480,13 @@
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
-				<xsd:attribute name="throw-exception-on-rejection" type="xsd:string" default="false" />
+				<xsd:attribute name="throw-exception-on-rejection" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+						Throw an exception if the filter rejects the message (default false).
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>
@@ -3105,7 +3111,7 @@ is provided, the return value is expected to match a channel name exactly.
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="ignore-send-failures" default="false">
+		<xsd:attribute name="ignore-send-failures">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
 					If set to "true", failures to send to a message channel will
@@ -3124,7 +3130,7 @@ is provided, the return value is expected to match a channel name exactly.
                 <xsd:union memberTypes="xsd:boolean xsd:string" />
             </xsd:simpleType>
 		</xsd:attribute>
-		<xsd:attribute name="apply-sequence" default="false">
+		<xsd:attribute name="apply-sequence">
 			<xsd:annotation>
 				<xsd:documentation>
 					Specify whether sequence number and size headers should be added to each
@@ -3135,7 +3141,7 @@ is provided, the return value is expected to match a channel name exactly.
                 <xsd:union memberTypes="xsd:boolean xsd:string" />
             </xsd:simpleType>
 		</xsd:attribute>
-        <xsd:attribute name="resolution-required" default="true">
+        <xsd:attribute name="resolution-required">
             <xsd:annotation>
                 <xsd:documentation>
                     Specify whether channel names must always be successfully resolved

spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelegatingConsumerParserTests-context.xml

@@ -0,0 +1,105 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<int:channel id="in" />
+
+	<int:channel id="discard">
+		<int:queue />
+	</int:channel>
+
+	<!-- Filters -->
+
+	<int:filter id="directFilter" input-channel="in"
+			discard-channel="discard" send-timeout="123" throw-exception-on-rejection="true">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyFilter" />
+	</int:filter>
+
+	<int:filter id="refFilter" input-channel="in" ref="myFilter" />
+
+	<bean id="myFilter" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyFilter" />
+
+	<int:filter id="filterWithMessageSelectorThatsAlsoAnARPMH" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MySelectorShouldntBeUsedAsTheHandler" />
+	</int:filter>
+
+	<!-- Routers -->
+
+	<int:router id="directRouter" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyRouter" />
+	</int:router>
+
+	<int:router id="refRouter" input-channel="in" ref="myRouter" />
+
+	<bean id="myRouter" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyRouter" />
+
+	<int:router id="directRouterMH" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyRouterMH" />
+	</int:router>
+
+	<int:router id="refRouterMH" input-channel="in" ref="myRouterMH" />
+
+	<bean id="myRouterMH" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyRouterMH" />
+
+	<int:router id="directRouterARPMH" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyRouterARPMH" />
+	</int:router>
+
+	<int:router id="refRouterARPMH" input-channel="in" ref="myRouterARPMH" />
+
+	<bean id="myRouterARPMH" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyRouterARPMH" />
+
+	<!-- Service Activators -->
+
+	<int:service-activator id="directServiceARPMH" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyServiceARPMH" />
+	</int:service-activator>
+
+	<int:transformer id="refServiceARPMH" input-channel="in" ref="myServiceARPMH"/>
+
+	<bean id="myServiceARPMH" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyServiceARPMH" />
+
+	<!-- Splitters -->
+	
+	<int:splitter id="directSplitter" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MySplitter" />
+	</int:splitter>
+
+	<int:splitter id="refSplitter" input-channel="in" ref="mySplitter"/>
+
+	<bean id="mySplitter" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MySplitter" />
+
+	<int:splitter id="splitterWithARPMH" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MySplitterThatsAnARPMH" />
+	</int:splitter>
+
+	<int:splitter id="splitterWithARPMHWithAtts" input-channel="in" ref="splitterARPMH" send-timeout="123"/> 
+
+	<bean id="splitterARPMH" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MySplitterThatsAnARPMH" />
+
+<!-- 	<int:splitter id="splitterWithARPMHWithSplitterAtts" input-channel="in" ref="splitterARPMH2" method="handleMessage" apply-sequence="false"/>  -->
+
+<!-- 	<bean id="splitterARPMH2" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MySplitterThatsAnARPMH" /> -->
+
+	<!-- Transformers -->
+
+	<int:transformer id="directTransformer" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyTransformer" />
+	</int:transformer>
+
+	<int:transformer id="refTransformer" input-channel="in" ref="myTransformer" />
+
+	<bean id="myTransformer" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyTransformer" />
+
+	<int:transformer id="directTransformerARPMH" input-channel="in">
+		<bean class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyTransformerARPMH" />
+	</int:transformer>
+
+	<int:transformer id="refTransformerARPMH" input-channel="in" ref="myTransformerARPMH"/>
+
+	<bean id="myTransformerARPMH" class="org.springframework.integration.config.xml.DelegatingConsumerParserTests$MyTransformerARPMH" />
+
+</beans>

spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelegatingConsumerParserTests.java

@@ -0,0 +1,305 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.config.xml;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.ServiceActivatorFactoryBean;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.MessageSelector;
+import org.springframework.integration.filter.MessageFilter;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.router.AbstractMessageRouter;
+import org.springframework.integration.splitter.AbstractMessageSplitter;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.integration.transformer.AbstractTransformer;
+import org.springframework.integration.transformer.MessageTransformingHandler;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class DelegatingConsumerParserTests {
+
+	@Autowired @Qualifier("directFilter.handler")
+	private MessageHandler directFilter;
+
+	@Autowired @Qualifier("refFilter.handler")
+	private MessageHandler refFilter;
+
+	@Autowired @Qualifier("filterWithMessageSelectorThatsAlsoAnARPMH.handler")
+	private MessageHandler filterWithMessageSelectorThatsAlsoAnARPMH;
+
+	@Autowired @Qualifier("directRouter.handler")
+	private MessageHandler directRouter;
+
+	@Autowired @Qualifier("refRouter.handler")
+	private MessageHandler refRouter;
+
+	@Autowired @Qualifier("directRouterMH.handler")
+	private MessageHandler directRouterMH;
+
+	@Autowired @Qualifier("refRouterMH.handler")
+	private MessageHandler refRouterMH;
+
+	@Autowired @Qualifier("directRouterARPMH.handler")
+	private MessageHandler directRouterARPMH;
+
+	@Autowired @Qualifier("refRouterARPMH.handler")
+	private MessageHandler refRouterARPMH;
+
+	@Autowired @Qualifier("directServiceARPMH.handler")
+	private MessageHandler directServiceARPMH;
+
+	@Autowired @Qualifier("refServiceARPMH.handler")
+	private MessageHandler refServiceARPMH;
+
+	@Autowired @Qualifier("directSplitter.handler")
+	private MessageHandler directSplitter;
+
+	@Autowired @Qualifier("refSplitter.handler")
+	private MessageHandler refSplitter;
+
+	@Autowired @Qualifier("splitterWithARPMH.handler")
+	private MessageHandler splitterWithARPMH;
+
+	@Autowired @Qualifier("splitterWithARPMHWithAtts.handler")
+	private MessageHandler splitterWithARPMHWithAtts;
+
+	@Autowired @Qualifier("directTransformer.handler")
+	private MessageHandler directTransformer;
+
+	@Autowired @Qualifier("refTransformer.handler")
+	private MessageHandler refTransformer;
+
+	@Autowired @Qualifier("directTransformerARPMH.handler")
+	private MessageHandler directTransformerARPMH;
+
+	@Autowired @Qualifier("refTransformerARPMH.handler")
+	private MessageHandler refTransformerARPMH;
+
+	private static QueueChannel replyChannel = new QueueChannel();
+
+	@Test
+	public void testDelegates() {
+		assertTrue(directFilter instanceof MyFilter);
+		testHandler(directFilter);
+		assertTrue(refFilter instanceof MyFilter);
+		testHandler(refFilter);
+		// MessageSelector (wrapped in MessageFilter) wins here
+		assertTrue(filterWithMessageSelectorThatsAlsoAnARPMH instanceof MessageFilter);
+		testHandler(filterWithMessageSelectorThatsAlsoAnARPMH);
+
+		assertTrue(directRouter instanceof MyRouter);
+		testHandler(directRouter);
+		assertTrue(refRouter instanceof MyRouter);
+		testHandler(refRouter);
+		assertTrue(directRouterMH instanceof MyRouterMH);
+		testHandler(directRouterMH);
+		assertTrue(refRouterMH instanceof MyRouterMH);
+		testHandler(refRouterMH);
+		assertTrue(directRouterARPMH instanceof MyRouterARPMH);
+		testHandler(directRouterARPMH);
+		assertTrue(refRouterARPMH instanceof MyRouterARPMH);
+		testHandler(refRouterARPMH);
+
+		assertTrue(directServiceARPMH instanceof MyServiceARPMH);
+		testHandler(directServiceARPMH);
+		assertTrue(refServiceARPMH instanceof MyServiceARPMH);
+		testHandler(refServiceARPMH);
+
+		assertTrue(directSplitter instanceof MySplitter);
+		testHandler(directSplitter);
+		assertTrue(refSplitter instanceof MySplitter);
+		testHandler(refSplitter);
+		assertTrue(splitterWithARPMH instanceof MySplitterThatsAnARPMH);
+		testHandler(splitterWithARPMH);
+		assertTrue(splitterWithARPMHWithAtts instanceof MySplitterThatsAnARPMH);
+		assertEquals(Long.valueOf(123), TestUtils.getPropertyValue(splitterWithARPMHWithAtts, "messagingTemplate.sendTimeout", Long.class));
+		testHandler(splitterWithARPMHWithAtts);
+
+		assertTrue(directTransformer instanceof MessageTransformingHandler);
+		assertTrue(TestUtils.getPropertyValue(directTransformer, "transformer") instanceof MyTransformer);
+		testHandler(directTransformer);
+		assertTrue(refTransformer instanceof MessageTransformingHandler);
+		assertTrue(TestUtils.getPropertyValue(refTransformer, "transformer") instanceof MyTransformer);
+		testHandler(refTransformer);
+		assertTrue(directTransformerARPMH instanceof MyTransformerARPMH);
+		testHandler(directTransformerARPMH);
+		assertTrue(refTransformerARPMH instanceof MyTransformerARPMH);
+		testHandler(refTransformerARPMH);
+
+	}
+
+	@Test
+	public void testOneRefOnly() throws Exception {
+		ServiceActivatorFactoryBean fb = new ServiceActivatorFactoryBean();
+		fb.setBeanFactory(mock(BeanFactory.class));
+		MyServiceARPMH service = new MyServiceARPMH();
+		service.setBeanName("foo");
+		fb.setTargetObject(service);
+		fb.getObject();
+		fb = new ServiceActivatorFactoryBean();
+		fb.setBeanFactory(mock(BeanFactory.class));
+		fb.setTargetObject(service);
+		try {
+			fb.getObject();
+			fail("expected exception");
+		}
+		catch (Exception e) {
+			assertEquals("An AbstractReplyProducingMessageHandler may only be referenced once (foo) - "
+					+ "use scope=\"prototype\"", e.getMessage());
+		}
+	}
+
+	private void testHandler(MessageHandler handler) {
+		Message<?> message = MessageBuilder.withPayload("foo")
+				.setReplyChannel(replyChannel)
+				.build();
+		handler.handleMessage(message);
+		assertNotNull(replyChannel.receive(0));
+
+	}
+
+	public static class MyFilter extends MessageFilter {
+
+		public MyFilter() {
+			super(new MessageSelector() {
+
+				@Override
+				public boolean accept(Message<?> message) {
+					return true;
+				}
+			});
+		}
+
+	}
+
+	public static class MySelectorShouldntBeUsedAsTheHandler extends AbstractReplyProducingMessageHandler
+			implements MessageSelector {
+
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			return null;
+		}
+
+		@Override
+		public boolean accept(Message<?> message) {
+			return true;
+		}
+
+	}
+
+	public static class MyRouter extends AbstractMessageRouter {
+
+		@Override
+		protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
+			List<MessageChannel> channels = new ArrayList<MessageChannel>();
+			channels.add(replyChannel);
+			return channels;
+		}
+
+	}
+
+	public static class MyRouterMH implements MessageHandler {
+
+		@Override
+		public void handleMessage(Message<?> message) throws MessagingException {
+			replyChannel.send(message);
+		}
+
+	}
+
+	public static class MyRouterARPMH extends AbstractReplyProducingMessageHandler {
+
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			replyChannel.send(requestMessage);
+			return null;
+		}
+
+	}
+
+	public static class MyServiceARPMH extends AbstractReplyProducingMessageHandler {
+
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			return requestMessage;
+		}
+
+	}
+
+	public static class MyTransformer extends AbstractTransformer {
+
+		@Override
+		protected Object doTransform(Message<?> message) throws Exception {
+			return message;
+		}
+
+	}
+
+	public static class MyTransformerARPMH extends AbstractReplyProducingMessageHandler {
+
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			return requestMessage;
+		}
+
+	}
+
+	public static class MySplitter extends AbstractMessageSplitter {
+
+		@Override
+		protected Object splitMessage(Message<?> message) {
+			return message;
+		}
+
+	}
+
+	public static class MySplitterThatsAnARPMH extends AbstractReplyProducingMessageHandler {
+
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			return requestMessage;
+		}
+
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml

@@ -11,7 +11,32 @@
 
 	<service-activator id="gatewayTestService" input-channel="gatewayTestInputChannel" ref="gateway"/>
 
-	<service-activator id="handlerTestService" input-channel="handlerTestInputChannel" ref="testMessageHandler"/>
+	<service-activator id="replyingHandlerTestService" input-channel="replyingHandlerTestInputChannel">
+		<beans:bean
+			class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestReplyingMessageHandler"/>
+	</service-activator>
+
+	<service-activator id="optimizedRefReplyingHandlerTestService"
+		input-channel="optimizedRefReplyingHandlerTestInputChannel" ref="testReplyingMessageHandler"/>
+
+	<service-activator id="replyingHandlerWithStandardMethodTestService"
+		 input-channel="replyingHandlerWithStandardMethodTestInputChannel"
+		 method="handleMessage">
+		 <beans:bean
+			class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestReplyingMessageHandler"/>
+	</service-activator>
+
+	<service-activator id="replyingHandlerWithOtherMethodTestService"
+		 input-channel="replyingHandlerWithOtherMethodTestInputChannel"
+		 method="foo">
+		 <beans:bean
+			class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestReplyingMessageHandler"/>
+	</service-activator>
+
+	<service-activator id="handlerTestService" input-channel="handlerTestInputChannel">
+		<beans:bean
+			class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestMessageHandler"/>
+	</service-activator>
 
 	<service-activator id="processorTestService" input-channel="processorTestInputChannel" ref="testMessageProcessor"/>
 
@@ -25,7 +50,8 @@
 		<queue/>
 	</channel>
 
-	<beans:bean id="testMessageHandler" class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestMessageHandler"/>
+	<beans:bean id="testReplyingMessageHandler"
+		class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestReplyingMessageHandler"/>
 
 	<beans:bean id="testMessageProcessor" class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestMessageProcessor">
 		<beans:property name="prefix" value="foo"/>

spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -18,32 +18,27 @@ package org.springframework.integration.handler;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertThat;
 
-import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
-import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import java.util.Map;
-
 /**
  * See INT-1688 for background.
- * 
+ *
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Gary Russell
  * @since 2.0.1
  */
 @ContextConfiguration
@@ -53,6 +48,18 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
 	@Autowired
 	private MessageChannel gatewayTestInputChannel;
 
+	@Autowired
+	private MessageChannel replyingHandlerTestInputChannel;
+
+	@Autowired
+	private MessageChannel optimizedRefReplyingHandlerTestInputChannel;
+
+	@Autowired
+	private MessageChannel replyingHandlerWithStandardMethodTestInputChannel;
+
+	@Autowired
+	private MessageChannel replyingHandlerWithOtherMethodTestInputChannel;
+
 	@Autowired
 	private MessageChannel handlerTestInputChannel;
 
@@ -74,14 +81,58 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
 		assertEquals("gatewayTestInputChannel,gatewayTestService,gateway,requestChannel,bridge,replyChannel", reply.getHeaders().get("history").toString());
 	}
 
+	@Test
+	public void testReplyingMessageHandler() {
+		QueueChannel replyChannel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
+		this.replyingHandlerTestInputChannel.send(message);
+		Message<?> reply = replyChannel.receive(0);
+		assertEquals("TEST", reply.getPayload());
+		assertEquals("replyingHandlerTestInputChannel,replyingHandlerTestService", reply.getHeaders().get("history").toString());
+		StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
+		assertEquals("doDispatch", st[3].getMethodName()); // close to the metal
+	}
+
+	@Test
+	public void testNotOptimizedReplyingMessageHandler() {
+		QueueChannel replyChannel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
+		this.optimizedRefReplyingHandlerTestInputChannel.send(message);
+		Message<?> reply = replyChannel.receive(0);
+		assertEquals("TEST", reply.getPayload());
+		assertEquals("optimizedRefReplyingHandlerTestInputChannel,optimizedRefReplyingHandlerTestService",
+				reply.getHeaders().get("history").toString());
+		StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
+		assertEquals("doDispatch", st[3].getMethodName());
+	}
+
+	@Test
+	public void testReplyingMessageHandlerWithStandardMethod() {
+		QueueChannel replyChannel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
+		this.replyingHandlerWithStandardMethodTestInputChannel.send(message);
+		Message<?> reply = replyChannel.receive(0);
+		assertEquals("TEST", reply.getPayload());
+		assertEquals("replyingHandlerWithStandardMethodTestInputChannel,replyingHandlerWithStandardMethodTestService", reply.getHeaders().get("history").toString());
+		StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
+		assertEquals("doDispatch", st[3].getMethodName()); // close to the metal
+	}
+
+	@Test
+	public void testReplyingMessageHandlerWithOtherMethod() {
+		QueueChannel replyChannel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
+		this.replyingHandlerWithOtherMethodTestInputChannel.send(message);
+		Message<?> reply = replyChannel.receive(0);
+		assertEquals("bar", reply.getPayload());
+		assertEquals("replyingHandlerWithOtherMethodTestInputChannel,replyingHandlerWithOtherMethodTestService", reply.getHeaders().get("history").toString());
+	}
+
 	@Test
 	public void testMessageHandler() {
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
 		this.handlerTestInputChannel.send(message);
-		Message<?> reply = replyChannel.receive(0);
-		assertEquals("TEST", reply.getPayload());
-		assertEquals("handlerTestInputChannel,handlerTestService,testMessageHandler", reply.getHeaders().get("history").toString());
 	}
 
 //	INT-2399
@@ -100,19 +151,38 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
 
 
 	@SuppressWarnings("unused")
-	private static class TestMessageHandler extends AbstractReplyProducingMessageHandler {
+	private static class TestReplyingMessageHandler extends AbstractReplyProducingMessageHandler {
 
 		@Override
 		protected Object handleRequestMessage(Message<?> requestMessage) {
-			return requestMessage.getPayload().toString().toUpperCase();
+			Exception e = new RuntimeException();
+			StackTraceElement[] st = e.getStackTrace();
+			return MessageBuilder.withPayload(requestMessage.getPayload().toString().toUpperCase())
+					.setHeader("callStack", st);
 		}
+
+		public String foo(String in) {
+			return "bar";
+		}
+
 	}
 
+	@SuppressWarnings("unused")
+	private static class TestMessageHandler implements MessageHandler {
+
+		@Override
+		public void handleMessage(Message<?> requestMessage) {
+			Exception e = new RuntimeException();
+			StackTraceElement[] st = e.getStackTrace();
+			assertEquals("doDispatch", st[4].getMethodName());
+		}
+	}
 
 	private static class TestMessageProcessor implements MessageProcessor<String> {
 
 		private String prefix;
 
+		@SuppressWarnings("unused")
 		public void setPrefix(String prefix) {
 			this.prefix = prefix;
 		}

spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,13 +19,16 @@ package org.springframework.integration.transformer;
 import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
+
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
 import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.support.MessageBuilder;
 
 /**
  * @author Mark Fisher
@@ -39,12 +42,26 @@ public class TransformerContextTests {
 	public void methodInvokingTransformer() {
 		ApplicationContext context = new ClassPathXmlApplicationContext(
 				"transformerContextTests.xml", this.getClass());
-		MessageChannel input = (MessageChannel) context.getBean("input");
-		PollableChannel output = (PollableChannel) context.getBean("output");
+		MessageChannel input = context.getBean("input", MessageChannel.class);
+		PollableChannel output = context.getBean("output", PollableChannel.class);
 		input.send(new GenericMessage<String>("foo"));
 		Message<?> reply = output.receive(0);
 		assertEquals("FOO", reply.getPayload());
 		assertEquals(1, adviceCalled);
+
+		input = context.getBean("direct", MessageChannel.class);
+		input.send(new GenericMessage<String>("foo"));
+		reply = output.receive(0);
+		assertEquals("FOO", reply.getPayload());
+		StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
+		assertEquals("doDispatch", st[3].getMethodName()); // close to the metal
+
+		input = context.getBean("directRef", MessageChannel.class);
+		input.send(new GenericMessage<String>("foo"));
+		reply = output.receive(0);
+		assertEquals("FOO", reply.getPayload());
+		st = (StackTraceElement[]) reply.getHeaders().get("callStack");
+		assertEquals("doDispatch", st[3].getMethodName()); // SpEL
 	}
 
 	public static class FooAdvice extends AbstractRequestHandlerAdvice {
@@ -56,4 +73,16 @@ public class TransformerContextTests {
 		}
 
 	}
+
+	public static class Bar extends AbstractReplyProducingMessageHandler {
+
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			Exception e = new RuntimeException();
+			StackTraceElement[] st = e.getStackTrace();
+			return MessageBuilder.withPayload(requestMessage.getPayload().toString().toUpperCase())
+					.setHeader("callStack", st);
+		}
+
+	}
 }

spring-integration-core/src/test/java/org/springframework/integration/transformer/transformerContextTests.xml

@@ -20,5 +20,13 @@
 	</transformer>
 
 	<beans:bean id="testBean" class="org.springframework.integration.transformer.TestBean"/>
-
+
+	<transformer input-channel="direct" output-channel="output">
+		<beans:bean class="org.springframework.integration.transformer.TransformerContextTests$Bar"/>
+	</transformer>
+
+	<transformer input-channel="directRef" output-channel="output" ref="trans" method="handleMessage"/>
+	
+	<beans:bean id="trans" class="org.springframework.integration.transformer.TransformerContextTests$Bar"/>
+
 </beans:beans>