diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/DefaultHandlerEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/DefaultHandlerEndpointParser.java index ae2ee30545..62698b3846 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/DefaultHandlerEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/DefaultHandlerEndpointParser.java @@ -16,36 +16,17 @@ package org.springframework.integration.config; -import org.w3c.dom.Element; - -import org.springframework.beans.factory.config.RuntimeBeanReference; -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.handler.DefaultMessageHandlerAdapter; import org.springframework.integration.handler.MessageHandler; -import org.springframework.util.StringUtils; /** * @author Mark Fisher */ public class DefaultHandlerEndpointParser extends AbstractHandlerEndpointParser { - private static final String REPLY_HANDLER_ATTRIBUTE = "reply-handler"; - - private static final String REPLY_HANDLER_PROPERTY = "replyHandler"; - - @Override protected Class getHandlerAdapterClass() { return DefaultMessageHandlerAdapter.class; } - @Override - protected void postProcessEndpointBean(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) { - String replyHandler = element.getAttribute(REPLY_HANDLER_ATTRIBUTE); - if (StringUtils.hasText(replyHandler)) { - builder.addPropertyValue(REPLY_HANDLER_PROPERTY, new RuntimeBeanReference(replyHandler)); - } - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index a7081ef4f4..31ccfb5f91 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -203,19 +203,12 @@ - - - - + + + Defines an endpoint for a MessageHandler. - - - - - - - - + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java index 9213cff510..e8b65142db 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java @@ -20,7 +20,6 @@ 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.handler.ReplyHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryException; @@ -39,8 +38,6 @@ public class HandlerEndpoint extends AbstractEndpoint { private volatile MessageHandler handler; - private volatile ReplyHandler replyHandler = new EndpointReplyHandler(); - private volatile long replyTimeout = 1000; private volatile boolean returnAddressOverrides = false; @@ -56,11 +53,6 @@ public class HandlerEndpoint extends AbstractEndpoint { return this.handler; } - 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 @@ -126,33 +118,28 @@ public class HandlerEndpoint extends AbstractEndpoint { replyMessage = MessageBuilder.fromMessage(replyMessage) .setHeader(MessageHeaders.CORRELATION_ID, message.getId()).build(); } - this.replyHandler.handle(replyMessage, message); + if (replyMessage != null) { + MessageChannel replyChannel = resolveReplyChannel(replyMessage); + if (replyChannel == null) { + replyChannel = resolveReplyChannel(message); + if (replyChannel == null) { + throw new MessageHandlingException(replyMessage, "Unable to determine reply channel for message. " + + "Provide an 'outputChannelName' on the message endpoint or a 'returnAddress' in the message header"); + } + } + this.sendReplyMessage(replyMessage, replyChannel); + } } return null; } - - private class EndpointReplyHandler implements ReplyHandler { - - public void handle(Message replyMessage, Message originalMessage) { - if (replyMessage == null) { - return; - } - MessageChannel replyChannel = resolveReplyChannel(replyMessage); - if (replyChannel == null) { - replyChannel = resolveReplyChannel(originalMessage); - if (replyChannel == null) { - throw new MessageHandlingException(replyMessage, "Unable to determine reply channel for message. " + - "Provide an 'outputChannelName' on the message endpoint or a 'returnAddress' in the message header"); - } - } - if (logger.isDebugEnabled()) { - logger.debug("endpoint '" + HandlerEndpoint.this + "' replying to channel '" + replyChannel + "' with message: " + replyMessage); - } - if (!replyChannel.send(replyMessage, replyTimeout)) { - throw new MessageDeliveryException(replyMessage, - "unable to send reply message within alloted timeout of " + replyTimeout + " milliseconds"); - } + private void sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { + if (logger.isDebugEnabled()) { + logger.debug("endpoint '" + HandlerEndpoint.this + "' replying to channel '" + replyChannel + "' with message: " + replyMessage); + } + if (!replyChannel.send(replyMessage, replyTimeout)) { + throw new MessageDeliveryException(replyMessage, + "unable to send reply message within alloted timeout of " + replyTimeout + " milliseconds"); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java index 4111ea15a8..3ae9ad8f4a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java @@ -25,12 +25,11 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.RendezvousChannel; import org.springframework.integration.endpoint.EndpointRegistry; import org.springframework.integration.endpoint.HandlerEndpoint; -import org.springframework.integration.handler.ReplyHandler; import org.springframework.integration.handler.ReplyMessageCorrelator; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.MessageHeaders; +import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.selector.MessageSelector; @@ -155,10 +154,10 @@ public class RequestReplyTemplate implements MessageBusAware { } /** - * Send a request message whose reply should be handled be the provided callback. + * Send a request message whose reply should be sent to the provided target. */ - public boolean request(Message message, ReplyHandler replyHandler) { - MessageChannel replyChannelAdapter = new ReplyHandlingChannelAdapter(message, replyHandler); + public boolean request(Message message, MessageTarget target) { + MessageChannel replyChannelAdapter = new ReplyHandlingChannelAdapter(target); Message requestMessage = MessageBuilder.fromMessage(message) .setReturnAddress(replyChannelAdapter).build(); return this.send(requestMessage); @@ -228,14 +227,11 @@ public class RequestReplyTemplate implements MessageBusAware { private static class ReplyHandlingChannelAdapter implements MessageChannel { - private final Message originalMessage; - - private final ReplyHandler replyHandler; + private final MessageTarget target; - ReplyHandlingChannelAdapter(Message originalMessage, ReplyHandler replyHandler) { - this.originalMessage = originalMessage; - this.replyHandler = replyHandler; + ReplyHandlingChannelAdapter(MessageTarget target) { + this.target = target; } @@ -263,7 +259,7 @@ public class RequestReplyTemplate implements MessageBusAware { } public boolean send(Message message) { - this.replyHandler.handle(message, originalMessage); + this.target.send(message); return true; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyHandler.java deleted file mode 100644 index c4d6985aa3..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyHandler.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * 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; - -/** - * Strategy interface for handling reply messages. - * - * @author Mark Fisher - */ -public interface ReplyHandler { - - void handle(Message replyMessage, Message originalMessage); - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java index ad67332fb9..07ec1faff1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -103,18 +103,4 @@ public class EndpointParserTests { endpoint.send(message); } - @Test - public void testCustomReplyHandler() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointWithReplyHandler.xml", this.getClass()); - MessageTarget endpoint = (MessageTarget) context.getBean("endpoint"); - TestReplyHandler replyHandler = (TestReplyHandler) context.getBean("replyHandler"); - assertNull(replyHandler.getLastMessage()); - Message message = new StringMessage("test"); - endpoint.send(message); - Message reply = replyHandler.getLastMessage(); - assertNotNull(reply); - assertEquals("foo", reply.getPayload()); - } - } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestReplyHandler.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestReplyHandler.java deleted file mode 100644 index 9a742830f8..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestReplyHandler.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * 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.config; - -import org.springframework.integration.handler.ReplyHandler; -import org.springframework.integration.message.Message; - -/** - * @author Mark Fisher - */ -public class TestReplyHandler implements ReplyHandler { - - private volatile Message lastMessage; - - - public void handle(Message replyMessage, Message originalMessage) { - this.lastMessage = replyMessage; - } - - public Message getLastMessage() { - return this.lastMessage; - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithReplyHandler.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithReplyHandler.xml deleted file mode 100644 index 58394cf452..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithReplyHandler.xml +++ /dev/null @@ -1,22 +0,0 @@ - - - - - - - - - - - - - - - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/RequestReplyTemplateTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/RequestReplyTemplateTests.java index efc898d0db..558c42e1a1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/RequestReplyTemplateTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/RequestReplyTemplateTests.java @@ -30,8 +30,8 @@ import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.ReplyHandler; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.StringMessage; /** @@ -66,16 +66,17 @@ public class RequestReplyTemplateTests { public void testAsynchronousRequestAndReply() throws InterruptedException { final List replies = new ArrayList(3); final CountDownLatch latch = new CountDownLatch(3); - ReplyHandler replyHandler = new ReplyHandler() { - public void handle(Message replyMessage, Message originalMessage) { + MessageTarget replyTarget = new MessageTarget() { + public boolean send(Message replyMessage) { replies.add((String) replyMessage.getPayload()); latch.countDown(); + return true; } }; RequestReplyTemplate template = new RequestReplyTemplate(requestChannel); - template.request(new StringMessage("test1"), replyHandler); - template.request(new StringMessage("test2"), replyHandler); - template.request(new StringMessage("test3"), replyHandler); + template.request(new StringMessage("test1"), replyTarget); + template.request(new StringMessage("test2"), replyTarget); + template.request(new StringMessage("test3"), replyTarget); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertTrue(replies.contains("TEST1"));