diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index ffc7e3db30..cfe919d5b1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/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++) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 3b19eda2f6..c11d61ae8d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -136,6 +136,7 @@ + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java index c0d8fbefcf..98fed0d331 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java +++ b/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); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java new file mode 100644 index 0000000000..25d5a12e28 --- /dev/null +++ b/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()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestBean.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestBean.java new file mode 100644 index 0000000000..6290c458ef --- /dev/null +++ b/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; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressIsFallbackByDefault.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressIsFallbackByDefault.xml new file mode 100644 index 0000000000..6f47dcea78 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressIsFallbackByDefault.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressOverrides.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressOverrides.xml new file mode 100644 index 0000000000..92e38855f6 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/returnAddressOverrides.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + +