Added "returnAddressOverrides" property for endpoints ('return-address-overrides' in XML). The default value is 'false' indicating that any explicitly defined output channel should take precedence.

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

@@ -53,6 +53,8 @@ public class EndpointParser implements BeanDefinitionParser {
 
 	private static final String DEFAULT_OUTPUT_CHANNEL_PROPERTY = "defaultOutputChannelName";
 
+	private static final String RETURN_ADDRESS_OVERRIDES_ATTRIBUTE = "return-address-overrides";
+
 	private static final String SELECTOR_ELEMENT = "selector";
 
 	private static final String SELECTORS_PROPERTY = "messageSelectors";
@@ -97,6 +99,9 @@ public class EndpointParser implements BeanDefinitionParser {
 		if (StringUtils.hasText(defaultOutputChannel)) {
 			endpointDef.getPropertyValues().addPropertyValue(DEFAULT_OUTPUT_CHANNEL_PROPERTY, defaultOutputChannel);
 		}
+		String returnAddressOverridesAttribute = element.getAttribute(RETURN_ADDRESS_OVERRIDES_ATTRIBUTE);
+		boolean returnAddressOverrides = "true".equals(returnAddressOverridesAttribute);
+		endpointDef.getPropertyValues().addPropertyValue("returnAddressOverrides", returnAddressOverrides);
 		ManagedList selectors = new ManagedList();
 		NodeList childNodes = element.getChildNodes();
 		for (int i = 0; i < childNodes.getLength(); i++) {

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

@@ -136,6 +136,7 @@
 					<xsd:attribute name="handler" type="xsd:string" use="required"/>
 					<xsd:attribute name="handler-method" type="xsd:string"/>
 					<xsd:attribute name="reply-handler" type="xsd:string"/>
+					<xsd:attribute name="return-address-overrides" type="xsd:boolean" default="false"/>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java

@@ -45,6 +45,8 @@ public class HandlerEndpoint extends TargetEndpoint {
 
 	private volatile String defaultOutputChannelName;
 
+	private volatile boolean returnAddressOverrides = false;
+
 
 	public HandlerEndpoint(MessageHandler handler) {
 		Assert.notNull(handler, "handler must not be null");
@@ -83,6 +85,10 @@ public class HandlerEndpoint extends TargetEndpoint {
 		return this.defaultOutputChannelName;
 	}
 
+	public void setReturnAddressOverrides(boolean returnAddressOverrides) {
+		this.returnAddressOverrides = returnAddressOverrides;
+	}
+
 	public void afterPropertiesSet() {
 		Assert.notNull(this.handler, "handler must not be null");
 		if (this.handler instanceof ChannelRegistryAware) {
@@ -92,19 +98,42 @@ public class HandlerEndpoint extends TargetEndpoint {
 		super.afterPropertiesSet();
 	}
 
-
 	private MessageChannel resolveReplyChannel(MessageHeader originalMessageHeader) {
-		Object returnAddress = originalMessageHeader.getReturnAddress();
-		if (returnAddress instanceof MessageChannel) {
-			return (MessageChannel) returnAddress;
+		if (this.returnAddressOverrides) {
+			MessageChannel channel = this.getReturnAddress(originalMessageHeader);
+			if (channel == null) {
+				channel = this.getOutputChannel();
+			}
+			return channel;
 		}
-		ChannelRegistry registry = this.getChannelRegistry();
-		if (returnAddress instanceof String && registry != null) {
-			String channelName = (String) returnAddress;
-			if (StringUtils.hasText(channelName)) {
-				return registry.lookupChannel(channelName);
+		else {
+			MessageChannel channel = this.getOutputChannel();
+			if (channel == null) {
+				channel = this.getReturnAddress(originalMessageHeader);
+			}
+			return channel;
+		}
+	}
+
+	private MessageChannel getReturnAddress(MessageHeader originalMessageHeader) {
+		Object returnAddress = originalMessageHeader.getReturnAddress();
+		if (returnAddress != null) {
+			if (returnAddress instanceof MessageChannel) {
+				return (MessageChannel) returnAddress;
+			}
+			ChannelRegistry registry = this.getChannelRegistry();
+			if (returnAddress instanceof String && registry != null) {
+				String channelName = (String) returnAddress;
+				if (StringUtils.hasText(channelName)) {
+					return registry.lookupChannel(channelName);
+				}
 			}
 		}
+		return null;
+	}
+
+	private MessageChannel getOutputChannel() {
+		ChannelRegistry registry = this.getChannelRegistry();
 		if (this.defaultOutputChannelName != null && registry != null) {
 			return registry.lookupChannel(this.defaultOutputChannelName);
 		}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.endpoint;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.Test;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
+
+/**
+ * @author Mark Fisher
+ */
+public class ReturnAddressTests {
+
+	@Test
+	public void testReturnAddressOverrides() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"returnAddressOverrides.xml", this.getClass());
+		MessageChannel channel1 = (MessageChannel) context.getBean("channel1");
+		MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel");
+		context.start();
+		StringMessage message = new StringMessage("*");
+		message.getHeader().setReturnAddress("replyChannel");
+		channel1.send(message);
+		Message<?> response = replyChannel.receive(1000);
+		assertNotNull(response);
+		assertEquals("**", response.getPayload());
+	}
+
+	@Test
+	public void testReturnAddressIsFallbackByDefault() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"returnAddressIsFallbackByDefault.xml", this.getClass());
+		MessageChannel channel1 = (MessageChannel) context.getBean("channel1");
+		MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel");
+		context.start();
+		StringMessage message = new StringMessage("*");
+		message.getHeader().setReturnAddress("replyChannel");
+		channel1.send(message);
+		Message<?> response = replyChannel.receive(1000);
+		assertNotNull(response);
+		assertEquals("********", response.getPayload());
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestBean.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.endpoint;
+
+/**
+ * @author Mark Fisher
+ */
+public class TestBean {
+
+	public String duplicate(String input) {
+		return input + input;
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressIsFallbackByDefault.xml

@@ -0,0 +1,23 @@
+<?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:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<si:message-bus/>
+
+	<si:channel id="channel1"/>
+	<si:channel id="channel2"/>
+	<si:channel id="channel3"/>
+	<si:channel id="replyChannel"/>
+
+	<si:endpoint input-channel="channel1" handler="testBean" handler-method="duplicate" default-output-channel="channel2"/>
+	<si:endpoint input-channel="channel2" handler="testBean" handler-method="duplicate" default-output-channel="channel3"/>
+	<si:endpoint input-channel="channel3" handler="testBean" handler-method="duplicate"/>
+
+	<bean id="testBean" class="org.springframework.integration.endpoint.TestBean"/>
+
+</beans>

spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressOverrides.xml

@@ -0,0 +1,25 @@
+<?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:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<si:message-bus/>
+
+	<si:channel id="channel1"/>
+	<si:channel id="channel2"/>
+	<si:channel id="channel3"/>
+	<si:channel id="replyChannel"/>
+
+	<si:endpoint input-channel="channel1" handler="testBean" handler-method="duplicate"
+                 default-output-channel="channel2" return-address-overrides="true"/>
+
+	<si:endpoint input-channel="channel2" handler="testBean" handler-method="duplicate" default-output-channel="channel3"/>
+	<si:endpoint input-channel="channel3" handler="testBean" handler-method="duplicate"/>
+
+	<bean id="testBean" class="org.springframework.integration.endpoint.TestBean"/>
+
+</beans>