Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Removed ReplyHandler.

Browse Source
This commit is contained in:
Mark Fisher
2008-07-19 03:09:23 +00:00
parent 0aa8aeb129
commit fcc5d56f81
9 changed files with 38 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
org.springframework.integration/src/main/java/org/springframework/integration/config/DefaultHandlerEndpointParser.java
View File

@@ -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<? extends MessageHandler> 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));
}
}
}

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

@@ -203,19 +203,12 @@
</xsd:annotation>
</xsd:element>
<xsd:element name="handler-endpoint">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
<xsd:element name="handler-endpoint" type="inputOutputHandlerEndpointType">
<xsd:annotation>
<xsd:documentation>
Defines an endpoint for a MessageHandler.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="inputOutputHandlerEndpointType">
<xsd:attribute name="reply-handler" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:complexType name="inputOutputHandlerEndpointType">

49
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java
View File

@@ -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");
}
}

20
org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java
View File

@@ -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;
}

30
org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyHandler.java
View File

@@ -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);
}

14
org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java
View File

@@ -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());
}
}

38
org.springframework.integration/src/test/java/org/springframework/integration/config/TestReplyHandler.java
View File

@@ -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;
}
}

22
org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithReplyHandler.xml
View File

@@ -1,22 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
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">
<message-bus/>
<channel id="testChannel"/>
<handler-endpoint id="endpoint" input-channel="testChannel" ref="testHandler" reply-handler="replyHandler"/>
<beans:bean id="testHandler" class="org.springframework.integration.config.TestHandler">
<beans:property name="replyMessageText" value="foo"/>
</beans:bean>
<beans:bean id="replyHandler" class="org.springframework.integration.config.TestReplyHandler"/>
</beans:beans>

13
org.springframework.integration/src/test/java/org/springframework/integration/gateway/RequestReplyTemplateTests.java
View File

@@ -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<String> replies = new ArrayList<String>(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"));