Updated service-activator to use new SimpleEndpoint and DefaultMessageHandler. Modified EndpointInterceptor for preHandle/aroundHandle/postHandle with access-to and return-values-for the request/reply Messages.

org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java

@@ -332,8 +332,15 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 
 	private void activateEndpoint(MessageEndpoint endpoint) {
 		Assert.notNull(endpoint, "'endpoint' must not be null");
-		if (endpoint.getTarget() == null) {
-			this.lookupOrCreateChannel(endpoint.getOutputChannelName());
+		if (endpoint instanceof ChannelRegistryAware) {
+			((ChannelRegistryAware) endpoint).setChannelRegistry(this);
+		}
+		MessageTarget target = endpoint.getTarget();
+		if (target == null) {
+			target = this.lookupOrCreateChannel(endpoint.getOutputChannelName());
+			if (target != null) {
+				endpoint.setTarget(target);
+			}
 		}
 		MessageSource<?> source = endpoint.getSource();
 		if (source == null) {

org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.endpoint.HandlerEndpoint;
+import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
@@ -58,7 +59,7 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
 
 	@Override
 	protected Class<?> getBeanClass(Element element) {
-		return HandlerEndpoint.class;
+		return this.getEndpointClass();
 	}
 
 	@Override
@@ -116,8 +117,19 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
 		this.postProcessEndpointBean(builder, element, parserContext);
 	}
 
+	/**
+	 * Subclasses must override this to return the MessageHandler class.
+	 * @return
+	 */
 	protected abstract Class<? extends MessageHandler> getHandlerAdapterClass();
 
+	/**
+	 * Subclasses may override this to return a specific MessageEndpoint class.
+	 */
+	protected Class<? extends MessageEndpoint> getEndpointClass() {
+		return HandlerEndpoint.class;
+	}
+
 	protected void postProcessEndpointBean(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java

@@ -16,8 +16,10 @@
 
 package org.springframework.integration.config;
 
-import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
+import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.endpoint.SimpleEndpoint;
 import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.handler.DefaultMessageHandler;
 
 /**
  * Parser for the <service-activator> element.
@@ -26,9 +28,14 @@ import org.springframework.integration.handler.MessageHandler;
  */
 public class ServiceActivatorParser extends AbstractHandlerEndpointParser {
 
+	@Override
+	protected Class<? extends MessageEndpoint> getEndpointClass() {
+		return SimpleEndpoint.class;
+	}
+
 	@Override
 	protected Class<? extends MessageHandler> getHandlerAdapterClass() {
-		return DefaultMessageHandlerAdapter.class;
+		return DefaultMessageHandler.class;
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java

@@ -27,6 +27,7 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageExchangeTemplate;
 import org.springframework.integration.message.MessageRejectedException;
@@ -208,45 +209,41 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 		if (logger.isDebugEnabled()) {
 			logger.debug("endpoint '" + this + "' handling message: " + message);
 		}
-		return this.send(message, 0);
+		this.handleMessage(message, 0);
+		return true;
 	}
 
-	private boolean send(final Message<?> message, final int index) {
+	private Message<?> handleMessage(Message<?> message, final int index) {
 		if (index == 0) {
 			for (EndpointInterceptor interceptor : interceptors) {
-				if (!interceptor.preSend(message)) {
-					return false;
+				message = interceptor.preHandle(message);
+				if (message == null) {
+					return null;
 				}
 			}
 		}
 		if (index == interceptors.size()) {
-			boolean result = this.doSend(message);
+			if (!this.supports(message)) {
+				throw new MessageRejectedException(message, "unsupported message");
+			}
+			Message<?> reply = this.handleMessage(message);
 			for (int i = index - 1; i >= 0; i--) {
 				EndpointInterceptor interceptor = this.interceptors.get(i);
-				interceptor.postSend(message, result);
+				reply = interceptor.postHandle(message, reply);
 			}
-			return result;
+			if (reply != null) {
+				this.getMessageExchangeTemplate().send(message, this.target);
+			}
+			return null;
 		}
 		EndpointInterceptor nextInterceptor = interceptors.get(index);
-		return nextInterceptor.aroundSend(message, new MessageTarget() {
-			@SuppressWarnings("unchecked")
-			public boolean send(Message message) {
-				return AbstractEndpoint.this.send(message, index + 1);
+		return nextInterceptor.aroundHandle(message, new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				return AbstractEndpoint.this.handleMessage(message, index + 1);
 			}
 		});
 	}
 
-	private boolean doSend(Message<?> message) {
-		if (!this.supports(message)) {
-			throw new MessageRejectedException(message, "unsupported message");
-		}
-		Message<?> result = this.handleMessage(message);
-		if (result != null) {
-			return this.getMessageExchangeTemplate().send(message, this.target);
-		}
-		return true;
-	}
-
 	protected boolean supports(Message<?> message) {
 		if (this.selector != null && !this.selector.accept(message)) {
 			if (logger.isDebugEnabled()) {

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java

@@ -16,18 +16,18 @@
 
 package org.springframework.integration.endpoint;
 
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
 
 /**
  * @author Mark Fisher
  */
 public interface EndpointInterceptor {
 
-	boolean preSend(Message<?> message);
+	Message<?> preHandle(Message<?> requestMessage);
 
-	boolean aroundSend(Message<?> message, MessageTarget endpoint);
+	Message<?> aroundHandle(Message<?> message, MessageHandler handler);
 
-	void postSend(Message<?> message, boolean result);
+	Message<?> postHandle(Message<?> requestMessage, Message<?> replyMessage);
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java

@@ -17,7 +17,9 @@
 package org.springframework.integration.endpoint;
 
 import java.lang.reflect.Array;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -29,11 +31,16 @@ import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.CompositeMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageExchangeTemplate;
 import org.springframework.integration.message.MessageHandlingException;
 import org.springframework.integration.message.MessageHeaders;
+import org.springframework.integration.message.MessageRejectedException;
+import org.springframework.integration.message.MessageSource;
 import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.selector.MessageSelector;
+import org.springframework.integration.scheduling.Schedule;
 import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
@@ -59,7 +66,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class SimpleEndpoint<T extends MessageHandler> implements MessageTarget, BeanNameAware {
+public class SimpleEndpoint<T extends MessageHandler> implements MessageEndpoint, ChannelRegistryAware, BeanNameAware {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
@@ -75,6 +82,10 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageTarget,
 
 	private volatile boolean requiresReply = false;
 
+	private volatile MessageSelector selector;
+
+	private final List<EndpointInterceptor> interceptors = new ArrayList<EndpointInterceptor>();
+
 	private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
 
 
@@ -120,6 +131,21 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageTarget,
 		this.errorHandler = errorHandler;
 	}
 
+	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);
+		}
+	}
+
 	public void setChannelRegistry(ChannelRegistry channelRegistry) {
 		if (this.handler instanceof ChannelRegistryAware) {
 			((ChannelRegistryAware) this.handler).setChannelRegistry(channelRegistry);
@@ -149,9 +175,15 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageTarget,
 		this.messageExchangeTemplate.setSendTimeout(replyTimeout);
 	}
 
-	public boolean send(Message<?> requestMessage) {
+	public final boolean send(Message<?> requestMessage) {
+		if (requestMessage == null || requestMessage.getPayload() == null) {
+			throw new IllegalArgumentException("Message and its payload must not be null");
+		}
+		if (this.logger.isDebugEnabled()) {
+			this.logger.debug("endpoint '" + this + "' handling message: " + requestMessage);
+		}
 		try {
-			Message<?> replyMessage = this.handler.handle(requestMessage);
+			Message<?> replyMessage = this.handleMessage(requestMessage, 0);
 			if (!this.isValidReply(replyMessage)) {
 				if (this.requiresReply) {
 					throw new MessageHandlingException(requestMessage,
@@ -180,6 +212,48 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageTarget,
 		}
 	}
 
+	private Message<?> handleMessage(Message<?> requestMessage, final int index) {
+		if (requestMessage == null || requestMessage.getPayload() == null) {
+			return null;
+		}
+		if (index == 0) {
+			for (EndpointInterceptor interceptor : this.interceptors) {
+				requestMessage = interceptor.preHandle(requestMessage);
+				if (requestMessage == null) {
+					return null;
+				}
+			}
+		}
+		if (index == this.interceptors.size()) {
+			if (!this.supports(requestMessage)) {
+				throw new MessageRejectedException(requestMessage, "unsupported message");
+			}
+			Message<?> replyMessage = this.handler.handle(requestMessage);
+			for (int i = index - 1; i >= 0; i--) {
+				EndpointInterceptor interceptor = this.interceptors.get(i);
+				replyMessage = interceptor.postHandle(requestMessage, replyMessage);
+			}
+			return replyMessage;
+		}
+		EndpointInterceptor nextInterceptor = this.interceptors.get(index);
+		return nextInterceptor.aroundHandle(requestMessage, new MessageHandler() {
+			@SuppressWarnings("unchecked")
+			public Message<?> handle(Message message) {
+				return SimpleEndpoint.this.handleMessage(message, index + 1);
+			}
+		});
+	}
+
+	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;
+	}
+
 	private void sendReplyMessage(Message<?> replyMessage, Message<?> requestMessage) {
 		if (replyMessage == null) {
 			throw new MessageHandlingException(requestMessage, "reply message must not be null");
@@ -189,6 +263,9 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageTarget,
 			throw new MessageEndpointReplyException(replyMessage, requestMessage,
 					"unable to resolve reply target");
 		}
+		replyMessage = MessageBuilder.fromMessage(replyMessage)
+				.copyHeadersIfAbsent(requestMessage.getHeaders())
+				.setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestMessage.getHeaders().getId()).build();
 		if (!this.messageExchangeTemplate.send(replyMessage, replyTarget)) {
 			throw new MessageEndpointReplyException(replyMessage, requestMessage,
 					"failed to send reply to '" + replyTarget + "'");
@@ -249,4 +326,53 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageTarget,
 		return replyTarget;
 	}
 
+	public String toString() {
+		return (this.name != null) ? this.name : super.toString();
+	}
+
+	/* TODO: remove the following methods after they are removed from the MessageEndpoint interface. */
+
+	private String inputChannelName;
+	private String outputChannelName;
+	private MessageSource<?> source;
+
+	public String getInputChannelName() {
+		return this.inputChannelName;
+	}
+
+	public void setInputChannelName(String inputChannelName) {
+		this.inputChannelName = inputChannelName;
+	}
+
+	public String getOutputChannelName() {
+		return this.outputChannelName;
+	}
+
+	public void setOutputChannelName(String outputChannelName) {
+		this.outputChannelName = outputChannelName;
+	}
+
+	public void setReturnAddressOverrides(boolean b) {
+	}
+
+	public Schedule getSchedule() {
+		return null;
+	}
+
+	public MessageSource<?> getSource() {
+		return this.source;
+	}
+
+	public MessageTarget getTarget() {
+		return this.outputChannel;
+	}
+
+	public void setSource(MessageSource<?> source) {
+		this.source = source;
+	}
+
+	public void setTarget(MessageTarget target) {
+		this.outputChannel = (MessageChannel) target;
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java

@@ -17,8 +17,10 @@
 package org.springframework.integration.endpoint.interceptor;
 
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -36,9 +38,10 @@ import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.EndpointInterceptor;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
+import org.springframework.integration.message.AsyncMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
 import org.springframework.integration.util.ErrorHandler;
 import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
@@ -118,30 +121,36 @@ public class ConcurrencyInterceptor extends EndpointInterceptorAdapter
 	}
 
 	@Override
-	public boolean aroundSend(final Message<?> message, final MessageTarget endpoint) {
+	@SuppressWarnings("unchecked")
+	public Message<?> aroundHandle(final Message<?> requestMessage, final MessageHandler handler) {
 		try {
-			this.executor.execute(new Runnable() {
-				public void run() {
+			FutureTask<Message<?>> task = new FutureTask<Message<?>>(new Callable<Message<?>>() {
+				public Message<?> call() throws Exception {
 					try {
-						endpoint.send(message);
+						return handler.handle(requestMessage);
 					}
-					catch (Throwable t) {
+					catch (Exception e) {
 						if (logger.isDebugEnabled()) {
-							logger.debug("error occurred in handler execution", t);
+							logger.debug("error occurred in handler execution", e);
 						}
 						if (errorHandler != null) {
-							errorHandler.handle(t);
+							errorHandler.handle(e);
 						}
-						else if (logger.isWarnEnabled() && !logger.isDebugEnabled()) {
-							logger.warn("error occurred in handler execution", t);
+						else {
+							if (logger.isWarnEnabled() && !logger.isDebugEnabled()) {
+								logger.warn("error occurred in handler execution", e);
+							}
+							throw e;
 						}
 					}
+					return null;
 				}
-			});
-			return true;
+			});	
+			this.executor.execute(task);
+			return new AsyncMessage(task);
 		}
 		catch (RuntimeException e) {
-			throw new MessageHandlerRejectedExecutionException(message, e);
+			throw new MessageHandlerRejectedExecutionException(requestMessage, e);
 		}
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java

@@ -17,8 +17,8 @@
 package org.springframework.integration.endpoint.interceptor;
 
 import org.springframework.integration.endpoint.EndpointInterceptor;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
 
 /**
  * A convenience base class for implementing {@link EndpointInterceptor EndpointInterceptors}.
@@ -27,15 +27,16 @@ import org.springframework.integration.message.MessageTarget;
  */
 public class EndpointInterceptorAdapter implements EndpointInterceptor {
 
-	public boolean preSend(Message<?> message) {
-		return true;
+	public Message<?> preHandle(Message<?> requestMessage) {
+		return requestMessage;
 	}
 
-	public boolean aroundSend(Message<?> message, MessageTarget endpoint) {
-		return endpoint.send(message);
+	public Message<?> aroundHandle(Message<?> message, MessageHandler handler) {
+		return handler.handle(message);
 	}
 
-	public void postSend(Message<?> message, boolean result) {
+	public Message<?> postHandle(Message<?> requestMessage, Message<?> replyMessage) {
+		return replyMessage;
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java

@@ -22,8 +22,8 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.endpoint.EndpointInterceptor;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.TransactionCallback;
@@ -83,20 +83,18 @@ public class TransactionInterceptor extends EndpointInterceptorAdapter implement
 	}
 
 	@Override
-	public boolean aroundSend(final Message<?> message, final MessageTarget endpoint) {
+	public Message<?> aroundHandle(final Message<?> message, final MessageHandler handler) {
 		if (this.transactionTemplate == null) {
 			throw new ConfigurationException("TransactionInterceptor has not been initialized");
 		}
-		this.transactionTemplate.execute(new TransactionCallback() {
+		return (Message<?>) this.transactionTemplate.execute(new TransactionCallback() {
 			public Object doInTransaction(TransactionStatus status) {
 				if (logger.isDebugEnabled()) {
-					logger.debug("Executing endpoint '" + endpoint + "' within transaction [" + status + "]");
+					logger.debug("Invoking handler '" + handler + "' within transaction [" + status + "]");
 				}
-				endpoint.send(message);
-				return null;
+				return handler.handle(message);
             }
 		});
-		return true;
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java

@@ -0,0 +1,333 @@
+/*
+ * 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.handler;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.GenericTypeResolver;
+import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ParameterNameDiscoverer;
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.handler.annotation.Header;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.MessageHeaders;
+import org.springframework.integration.util.DefaultMethodInvoker;
+import org.springframework.integration.util.MethodInvoker;
+import org.springframework.integration.util.NameResolvingMethodInvoker;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * A base class for any {@link MessageHandler} that may act as an adapter
+ * by invoking a "plain" (not Message-aware) method for a given target object.
+ * When used as an adapter, the target Object is mandatory and either a
+ * {@link Method} reference or a 'methodName' must be provided. If no Object
+ * and Method are provided, the handler will simply process the request
+ * Message's payload.
+ * 
+ * @author Mark Fisher
+ */
+public abstract class AbstractMessageHandler implements MessageHandler, InitializingBean {
+
+	private static final Log logger = LogFactory.getLog(AbstractMessageHandler.class);
+
+	private volatile boolean methodExpectsMessage;
+
+	private volatile Object object;
+
+	private volatile Method method;
+
+	private volatile String methodName;
+
+	private volatile MethodInvoker invoker;
+
+	private volatile MethodParameterMetadata[] parameterMetadata;
+
+	private final ParameterNameDiscoverer parameterNameDiscoverer = new LocalVariableTableParameterNameDiscoverer();
+
+	private volatile boolean initialized;
+
+	private final Object initializationMonitor = new Object();
+
+
+	public AbstractMessageHandler(Object object, Method method) {
+		Assert.notNull(object, "object must not be null");
+		Assert.notNull(method, "method must not be null");
+		this.object = object;
+		this.method = method;
+		this.methodName = method.getName();
+	}
+
+	public AbstractMessageHandler(Object object, String methodName) {
+		Assert.notNull(object, "object must not be null");
+		Assert.notNull(methodName, "methodName must not be null");
+		this.object = object;
+		this.methodName = methodName;
+	}
+
+	public AbstractMessageHandler() {
+	}
+
+
+	public void setObject(Object object) {
+		this.object = object;
+	}
+
+	public void setMethod(Method method) {
+		Assert.notNull(method, "method must not be null");
+		this.method = method;
+		this.methodName = method.getName();
+	}
+
+	public void setMethodName(String methodName) {
+		Assert.notNull(methodName, "methodName must not be null");
+		if (this.method != null && !this.method.getName().equals(methodName)) {
+			this.method = null;
+		}
+		this.methodName = methodName;
+	}
+
+	public void afterPropertiesSet() {
+		synchronized (this.initializationMonitor) {
+			if (this.initialized) {
+				return;
+			}
+			if (this.object != null) {
+				if (this.method == null) {
+					if (this.methodName == null) {
+						throw new IllegalStateException("either the 'method' or 'methodName' must be provided when the 'object' property has been set");
+					}
+					final List<Method> candidates = new ArrayList<Method>();
+					ReflectionUtils.doWithMethods(this.object.getClass(), new ReflectionUtils.MethodCallback() {
+						public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+							if (method.getName().equals(AbstractMessageHandler.this.methodName)) {
+								candidates.add(method);
+							}
+						}
+					});
+					if (candidates.size() == 0) {
+						throw new ConfigurationException("no such method '" + this.methodName
+								+ "' on target class [" + this.object.getClass() + "]"); 
+					}
+					if (candidates.size() == 1) {
+						this.method = candidates.get(0);
+					}
+				}
+				if (this.method != null) {
+					this.invoker = new DefaultMethodInvoker(this.object, this.method);
+				}
+				else {
+					// TODO: resolve the candidate method and/or create a dynamic resolver
+					this.invoker = new NameResolvingMethodInvoker(this.object, this.methodName);
+				}
+				this.configureParameterMetadata();
+			}
+			this.initialized = true;
+		}
+	}
+
+	private void configureParameterMetadata() {
+		if (this.method == null) {
+			return;
+		}
+		Class<?>[] paramTypes = this.method.getParameterTypes();			
+		this.parameterMetadata = new MethodParameterMetadata[paramTypes.length];
+		for (int i = 0; i < parameterMetadata.length; i++) {
+			MethodParameter methodParam = new MethodParameter(this.method, i);
+			methodParam.initParameterNameDiscovery(this.parameterNameDiscoverer);
+			GenericTypeResolver.resolveParameterType(methodParam, this.method.getDeclaringClass());
+			Object[] paramAnnotations = methodParam.getParameterAnnotations();
+			String headerName = null;
+			for (int j = 0; j < paramAnnotations.length; j++) {
+				if (Header.class.isInstance(paramAnnotations[j])) {
+					Header headerAnnotation = (Header) paramAnnotations[j];
+					headerName = this.resolveParameterNameIfNecessary(headerAnnotation.value(), methodParam);
+					parameterMetadata[i] = new MethodParameterMetadata(Header.class, headerName, headerAnnotation.required());
+				}
+			}
+			if (headerName == null) {
+				parameterMetadata[i] = new MethodParameterMetadata(methodParam.getParameterType(), null, false);
+			}
+		}
+	}
+
+	public Message<?> handle(Message<?> message) {
+		if (message == null || message.getPayload() == null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("message handler received a null message");
+			}
+			return null;
+		}
+		if (!this.initialized) {
+			this.afterPropertiesSet();
+		}
+		Object result = (this.invoker != null) ? this.invokeHandlerMethod(message) : message.getPayload();
+		if (result == null) {
+			return null;
+		}
+		return this.createReplyMessage(result, message.getHeaders());
+	}
+
+	/**
+	 * Subclasses must implement this method to generate the reply Message.
+	 * 
+	 * @param result the return value from an adapter method, or the Message payload if not acting as an adapter
+	 * @param requestHeaders the MessageHeaders of the original request Message
+	 * @return the Message to be sent to the reply MessageTarget
+	 */
+	protected abstract Message<?> createReplyMessage(Object result, MessageHeaders requestHeaders);
+
+
+	private Object invokeHandlerMethod(Message<?> message) {
+		if (this.invoker == null) {
+			throw new IllegalStateException("cannot invoke method, invoker is null");
+		}
+		Object args[] = null;
+		Object mappingResult = this.methodExpectsMessage ? message
+				: this.mapMessageToMethodArguments(message);
+		if (mappingResult != null && mappingResult.getClass().isArray()
+				&& (Object.class.isAssignableFrom(mappingResult.getClass().getComponentType()))) {
+			args = (Object[]) mappingResult;
+		}
+		else {
+			args = new Object[] { mappingResult }; 
+		}
+		try {
+			Object result = null;
+			try {
+				result = this.invoker.invokeMethod(args);
+			}
+			catch (NoSuchMethodException e) {
+				try {
+					result = this.invoker.invokeMethod(args);
+					this.methodExpectsMessage = true;
+				}
+				catch (NoSuchMethodException e2) {
+					throw new MessageHandlingException(message, "unable to determine method match");
+				}
+			}
+			if (result == null) {
+				return null;
+			}
+			return result;
+		}
+		catch (InvocationTargetException e) {
+			throw new MessageHandlingException(message, "Handler method '"
+					+ this.methodName + "' threw an Exception.", e.getTargetException());
+		}
+		catch (Throwable e) {
+			throw new MessageHandlingException(message, "Failed to invoke handler method '"
+					+ this.methodName + "' with arguments: " + ObjectUtils.nullSafeToString(args), e);
+		}
+	}
+
+
+	private Object[] mapMessageToMethodArguments(Message<?> message) {
+		if (message == null) {
+			return null;
+		}
+		if (message.getPayload() == null) {
+			throw new IllegalArgumentException("Message payload must not be null.");
+		}
+		if (ObjectUtils.isEmpty(this.parameterMetadata)) {
+			return new Object[] { message.getPayload() };
+		}
+		Object[] args = new Object[this.parameterMetadata.length];
+		for (int i = 0; i < this.parameterMetadata.length; i++) {
+			MethodParameterMetadata metadata = this.parameterMetadata[i];
+			Class<?> expectedType = metadata.type;
+			if (expectedType.equals(Header.class)) {
+				Object value = message.getHeaders().get(metadata.key);
+				if (value == null && metadata.required) {
+					throw new MessageHandlingException(message,
+							"required header '" + metadata.key + "' not available");
+				}
+				args[i] = value;
+			}
+			else if (expectedType.isAssignableFrom(message.getClass())) {
+				args[i] = message;
+			}
+			else if (expectedType.isAssignableFrom(message.getPayload().getClass())) {
+				args[i] = message.getPayload();
+			}
+			else if (expectedType.equals(Map.class)) {
+				args[i] = message.getHeaders();
+			}
+			else if (expectedType.equals(Properties.class)) {
+				args[i] = this.getStringTypedHeaders(message);
+			}
+			else {
+				args[i] = message.getPayload();
+			}
+		}
+		return args;
+	}
+
+	private Properties getStringTypedHeaders(Message<?> message) {
+		Properties properties = new Properties();
+		MessageHeaders headers = message.getHeaders();
+		for (String key : headers.keySet()) {
+			Object value = headers.get(key);
+			if (value instanceof String) {
+				properties.setProperty(key, (String) value);
+			}
+		}
+		return properties;
+	}
+
+	private String resolveParameterNameIfNecessary(String paramName, MethodParameter methodParam) {
+		if (!StringUtils.hasText(paramName)) {
+			paramName = methodParam.getParameterName();
+			if (paramName == null) {
+				throw new IllegalStateException("No parameter name specified and not available in class file.");
+			}
+		}
+		return paramName;
+	}
+
+
+	private static class MethodParameterMetadata {
+
+		private final Class<?> type;
+
+		private final String key;
+
+		private final boolean required;
+
+
+		MethodParameterMetadata(Class<?> type, String key, boolean required) {
+			this.type = type;
+			this.key = key;
+			this.required = required;
+		}
+
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/handler/DefaultMessageHandler.java

@@ -0,0 +1,37 @@
+/*
+ * 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.handler;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageHeaders;
+
+/**
+ * The default MessageHandler implementation. Creates a Message for the reply payload.
+ * The request Message's headers are copied and the request Message's ID is set as this
+ * reply Message's correlationId.
+ * 
+ * @author Mark Fisher
+ */
+public class DefaultMessageHandler extends AbstractMessageHandler {
+
+	@Override
+	protected Message<?> createReplyMessage(Object result, MessageHeaders requestHeaders) {
+		return MessageBuilder.fromPayload(result).copyHeaders(requestHeaders).setCorrelationId(requestHeaders.getId()).build();
+	}
+
+}

org.springframework.integration/src/test/java/org/springframework/integration/config/TestAroundSendEndpointInterceptor.java

@@ -19,8 +19,8 @@ package org.springframework.integration.config;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.integration.endpoint.interceptor.EndpointInterceptorAdapter;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
 
 /**
  * @author Mark Fisher
@@ -35,11 +35,11 @@ public class TestAroundSendEndpointInterceptor extends EndpointInterceptorAdapte
 	}
 
 	@Override
-	public boolean aroundSend(Message<?> message, MessageTarget endpoint) {
+	public Message<?> aroundHandle(Message<?> message, MessageHandler handler) {
 		this.counter.incrementAndGet();
-		boolean result = endpoint.send(message);
+		Message<?> reply = handler.handle(message);
 		this.counter.incrementAndGet();
-		return result;
+		return reply;
 	}
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/config/TestPreSendInterceptor.java

@@ -34,9 +34,9 @@ public class TestPreSendInterceptor extends EndpointInterceptorAdapter {
 	}
 
 	@Override
-	public boolean preSend(Message<?> message) {
+	public Message<?> preHandle(Message<?> message) {
 		this.counter.incrementAndGet();
-		return true;
+		return message;
 	}
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.endpoint;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 import org.junit.Test;
 
@@ -34,17 +35,19 @@ import org.springframework.integration.message.StringMessage;
 public class ReturnAddressTests {
 
 	@Test
-	public void testReturnAddressOverrides() {
+	public void testNextTargetOverrides() {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"returnAddressTests.xml", this.getClass());
 		MessageChannel channel1 = (MessageChannel) context.getBean("channel1WithOverride");
 		PollableChannel replyChannel = (PollableChannel) context.getBean("replyChannel");
 		context.start();
 		Message<String> message = MessageBuilder.fromPayload("*")
-				.setReturnAddress("replyChannel").build();
+				.setNextTarget("replyChannel").build();
 		channel1.send(message);
 		Message<?> response = replyChannel.receive(3000);
 		assertNotNull(response);
+		PollableChannel outputChannel = (PollableChannel) context.getBean("channel2");
+		assertNull(outputChannel.receive(0));
 		assertEquals("**", response.getPayload());
 	}
 

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java

@@ -27,6 +27,7 @@ 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;
@@ -156,11 +157,16 @@ public class TransactionInterceptorTests {
 	}
 
 	@Test(expected = IllegalTransactionStateException.class)
-	public void testPropagationMandatoryCalledWithoutTransaction() {
+	public void testPropagationMandatoryCalledWithoutTransaction() throws Throwable {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"transactionInterceptorPropagationTests.xml", this.getClass());
 		final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("mandatory");
-		endpoint.send(new StringMessage("test"));
+		try {
+			endpoint.send(new StringMessage("test"));
+		}
+		catch (MessageHandlingException e) {
+			throw e.getCause();
+		}
 	}
 
 }