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 2f07a0cc0e..f3a2fafbb2 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 @@ -80,6 +80,10 @@ public class EndpointParser implements BeanDefinitionParser { private static final String ERROR_HANDLER_PROPERTY = "errorHandler"; + private static final String REPLY_HANDLER_ATTRIBUTE = "reply-handler"; + + private static final String REPLY_HANDLER_PROPERTY = "replyHandler"; + private static final String OBJECT_PROPERTY = "object"; private static final String METHOD_NAME_PROPERTY = "methodName"; @@ -178,6 +182,10 @@ public class EndpointParser implements BeanDefinitionParser { if (StringUtils.hasText(errorHandlerRef)) { endpointDef.getPropertyValues().addPropertyValue(ERROR_HANDLER_PROPERTY, new RuntimeBeanReference(errorHandlerRef)); } + String replyHandlerRef = element.getAttribute(REPLY_HANDLER_ATTRIBUTE); + if (StringUtils.hasText(replyHandlerRef)) { + endpointDef.getPropertyValues().addPropertyValue(REPLY_HANDLER_PROPERTY, new RuntimeBeanReference(replyHandlerRef)); + } String beanName = element.getAttribute(ID_ATTRIBUTE); if (!StringUtils.hasText(beanName)) { beanName = parserContext.getReaderContext().generateBeanName(endpointDef); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index d1e4172e59..f142f7480a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -128,6 +128,7 @@ + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index c54428243c..c0882be442 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -71,7 +71,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA private final List selectors = new CopyOnWriteArrayList(); - private final ReplyHandler replyHandler = new EndpointReplyHandler(); + private volatile ReplyHandler replyHandler = new EndpointReplyHandler(); private volatile long replyTimeout = 1000; @@ -149,6 +149,11 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA return (this.errorHandler != null); } + public void setReplyHandler(ReplyHandler replyHandler) { + Assert.notNull(replyHandler, "'replyHandler' must not be null"); + this.replyHandler = replyHandler; + } + /** * Set the timeout in milliseconds to be enforced when this endpoint sends a * reply message. If the message is not sent successfully within the diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 68feeee6cd..c52caf2cf5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -157,4 +157,18 @@ public class EndpointParserTests { assertEquals(message, exception.getFailedMessage()); } + @Test + public void testCustomReplyHandler() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "endpointWithReplyHandler.xml", this.getClass()); + MessageHandler endpoint = (MessageHandler) context.getBean("endpoint"); + TestReplyHandler replyHandler = (TestReplyHandler) context.getBean("replyHandler"); + assertNull(replyHandler.getLastMessage()); + Message message = new StringMessage("test"); + endpoint.handle(message); + Message reply = replyHandler.getLastMessage(); + assertNotNull(reply); + assertEquals("foo", reply.getPayload()); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java index 32459ea1b0..642fcf995e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java @@ -34,6 +34,10 @@ public class TestHandler implements MessageHandler { private String replyMessageText = null; + public TestHandler() { + this(1); + } + public TestHandler(int countdown) { this.latch = new CountDownLatch(countdown); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestReplyHandler.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestReplyHandler.java new file mode 100644 index 0000000000..5818a00e24 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestReplyHandler.java @@ -0,0 +1,39 @@ +/* + * Copyright 2002-2007 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; + +import org.springframework.integration.handler.ReplyHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHeader; + +/** + * @author Mark Fisher + */ +public class TestReplyHandler implements ReplyHandler { + + private volatile Message lastMessage; + + + public void handle(Message replyMessage, MessageHeader originalMessageHeader) { + this.lastMessage = replyMessage; + } + + public Message getLastMessage() { + return this.lastMessage; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithReplyHandler.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithReplyHandler.xml new file mode 100644 index 0000000000..f7781e4cef --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithReplyHandler.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + + + +