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

Removed the HandlerEndpoint (now replaced by SimpleEndpoint).

Browse Source
This commit is contained in:
Mark Fisher
2008-08-13 15:49:24 +00:00
parent a3f6254856
commit e16dce8aba
6 changed files with 35 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,150 +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.endpoint;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Implementation of the {@link MessageEndpoint} interface for invoking
* {@link MessageHandler MessageHandlers}.
*
* @author Mark Fisher
*/
public class HandlerEndpoint extends AbstractEndpoint {
private volatile MessageHandler handler;
private volatile long replyTimeout = 1000;
private volatile boolean returnAddressOverrides = false;
public HandlerEndpoint(MessageHandler handler) {
Assert.notNull(handler, "handler must not be null");
this.handler = handler;
}
public MessageHandler getHandler() {
return this.handler;
}
/**
* Set the timeout in milliseconds to be enforced when this endpoint sends a
* reply message. If the message is not sent successfully within the
* allotted time, then a MessageDeliveryException will be thrown.
* The default <code>replyTimeout</code> value is 1000 milliseconds.
*/
public void setReplyTimeout(long replyTimeout) {
this.replyTimeout = replyTimeout;
}
public void setReturnAddressOverrides(boolean returnAddressOverrides) {
this.returnAddressOverrides = returnAddressOverrides;
}
public void initialize() {
Assert.notNull(this.handler, "handler must not be null");
if (this.handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.handler).setChannelRegistry(this.getChannelRegistry());
}
super.initialize();
}
private MessageTarget resolveReturnAddress(Message<?> originalMessage) {
if (this.returnAddressOverrides) {
MessageTarget target = this.getReturnAddress(originalMessage);
if (target == null) {
target = this.getTarget();
}
return target;
}
else {
MessageTarget target = this.getTarget();
if (target == null) {
target = this.getReturnAddress(originalMessage);
}
return target;
}
}
private MessageTarget getReturnAddress(Message<?> originalMessage) {
Object returnAddress = originalMessage.getHeaders().getReturnAddress();
if (returnAddress != null) {
if (returnAddress instanceof MessageTarget) {
return (MessageTarget) 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;
}
@Override
protected Message<?> handleMessage(Message<?> message) {
Message<?> replyMessage = this.handler.handle(message);
if (replyMessage != null) {
Object correlationId = replyMessage.getHeaders().getCorrelationId();
if (correlationId == null) {
replyMessage = MessageBuilder.fromMessage(replyMessage)
.setHeader(MessageHeaders.CORRELATION_ID, message.getHeaders().getId()).build();
}
if (replyMessage != null) {
MessageTarget returnAddress = resolveReturnAddress(replyMessage);
if (returnAddress == null) {
returnAddress = resolveReturnAddress(message);
if (returnAddress == null) {
throw new MessageHandlingException(replyMessage, "Unable to determine return address for message. " +
"Provide an 'outputChannelName' on the message endpoint or a 'returnAddress' in the message header");
}
}
this.sendReplyMessage(replyMessage, returnAddress);
}
}
return null;
}
private void sendReplyMessage(Message<?> replyMessage, MessageTarget returnAddress) {
if (logger.isDebugEnabled()) {
logger.debug("endpoint '" + HandlerEndpoint.this + "' replying to target '" + returnAddress + "' with message: " + replyMessage);
}
boolean sent = (this.replyTimeout >= 0 && returnAddress instanceof BlockingTarget)
? ((BlockingTarget) returnAddress).send(replyMessage, this.replyTimeout)
: returnAddress.send(replyMessage);
if (!sent) {
throw new MessageDeliveryException(replyMessage,
"unable to send reply message within allotted timeout of " + this.replyTimeout + " milliseconds");
}
}
}

5
org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java
View File

@@ -33,7 +33,6 @@ import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.PollableChannelAdapter;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dispatcher.PublishSubscribeChannel;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.ErrorMessage;
@@ -219,8 +218,8 @@ public class DefaultMessageBusTests {
return message;
}
};
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setName("testEndpoint");
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setBeanName("testEndpoint");
endpoint.setSource(channelAdapter);
bus.registerEndpoint(endpoint);
bus.start();

10
org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -28,7 +28,7 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ThreadLocalChannel;
import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
@@ -55,10 +55,10 @@ public class DirectChannelSubscriptionTests {
@Test
public void testSendAndReceiveForRegisteredEndpoint() {
HandlerEndpoint endpoint = new HandlerEndpoint(new TestHandler());
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(new TestHandler());
endpoint.setInputChannelName("sourceChannel");
endpoint.setOutputChannelName("targetChannel");
endpoint.setName("testEndpoint");
endpoint.setBeanName("testEndpoint");
bus.registerEndpoint(endpoint);
bus.start();
this.sourceChannel.send(new StringMessage("foo"));
@@ -84,14 +84,14 @@ public class DirectChannelSubscriptionTests {
public void testExceptionThrownFromRegisteredEndpoint() {
QueueChannel errorChannel = new QueueChannel();
bus.setErrorChannel(errorChannel);
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(new MessageHandler() {
public Message<?> handle(Message<?> message) {
throw new RuntimeException("intentional test failure");
}
});
endpoint.setInputChannelName("sourceChannel");
endpoint.setOutputChannelName("targetChannel");
endpoint.setName("testEndpoint");
endpoint.setBeanName("testEndpoint");
bus.registerEndpoint(endpoint);
bus.start();
this.sourceChannel.send(new StringMessage("foo"));

2
org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml
View File

@@ -10,7 +10,7 @@
<bean id="targetChannel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="endpoint" class="org.springframework.integration.endpoint.HandlerEndpoint">
<bean id="endpoint" class="org.springframework.integration.endpoint.SimpleEndpoint">
<constructor-arg ref="handler"/>
<property name="source" ref="sourceChannel"/>
<property name="outputChannelName" value="targetChannel"/>

25
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -25,7 +25,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.message.Message;
@@ -73,9 +74,9 @@ public class SimpleDispatcherTests {
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final AtomicInteger selectorCounter = new AtomicInteger();
HandlerEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
HandlerEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
HandlerEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
SimpleEndpoint<?> endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
SimpleEndpoint<?> endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
SimpleEndpoint<?> endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
endpoint1.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint2.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint3.setSelector(new TestMessageSelector(selectorCounter, true));
@@ -100,9 +101,9 @@ public class SimpleDispatcherTests {
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final AtomicInteger selectorCounter = new AtomicInteger();
HandlerEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
HandlerEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
HandlerEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
SimpleEndpoint<?> endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
SimpleEndpoint<?> endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
SimpleEndpoint<?> endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
endpoint1.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint2.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint3.setSelector(new TestMessageSelector(selectorCounter, false));
@@ -132,9 +133,9 @@ public class SimpleDispatcherTests {
TestMessageHandler handler1 = new TestMessageHandler(handlerCounter, 4);
TestMessageHandler handler2 = new TestMessageHandler(handlerCounter, 4);
TestMessageHandler handler3 = new TestMessageHandler(handlerCounter, 2);
HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
HandlerEndpoint endpoint3 = new HandlerEndpoint(handler3);
SimpleEndpoint<?> endpoint1 = new SimpleEndpoint<MessageHandler>(handler1);
SimpleEndpoint<?> endpoint2 = new SimpleEndpoint<MessageHandler>(handler2);
SimpleEndpoint<?> endpoint3 = new SimpleEndpoint<MessageHandler>(handler3);
dispatcher.addTarget(endpoint1);
dispatcher.addTarget(endpoint2);
dispatcher.addTarget(endpoint3);
@@ -146,8 +147,8 @@ public class SimpleDispatcherTests {
}
private static HandlerEndpoint createEndpoint(MessageHandler handler) {
return new HandlerEndpoint(handler);
private static SimpleEndpoint<MessageHandler> createEndpoint(MessageHandler handler) {
return new SimpleEndpoint<MessageHandler>(handler);
}

28
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java
View File

@@ -46,7 +46,7 @@ import org.springframework.integration.message.selector.MessageSelectorChain;
public class HandlerEndpointTests {
@Test
public void testDefaultReplyChannel() throws Exception {
public void testOutputChannel() throws Exception {
QueueChannel replyChannel = new QueueChannel();
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel("replyChannel", replyChannel);
@@ -55,9 +55,9 @@ public class HandlerEndpointTests {
return new StringMessage("hello " + message.getPayload());
}
};
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setOutputChannelName("replyChannel");
endpoint.setTarget(replyChannel);
endpoint.send(new StringMessage("test"));
Message<?> reply = replyChannel.receive(50);
assertNotNull(reply);
@@ -72,7 +72,7 @@ public class HandlerEndpointTests {
return new StringMessage("hello " + message.getPayload());
}
};
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
Message<String> testMessage = MessageBuilder.fromPayload("test")
.setReturnAddress(replyChannel).build();
endpoint.send(testMessage);
@@ -91,7 +91,7 @@ public class HandlerEndpointTests {
return new StringMessage("hello " + message.getPayload());
}
};
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setChannelRegistry(channelRegistry);
Message<String> testMessage = MessageBuilder.fromPayload("test")
.setReturnAddress(replyChannel).build();
@@ -112,7 +112,7 @@ public class HandlerEndpointTests {
return new StringMessage("hello " + message.getPayload());
}
};
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setChannelRegistry(channelRegistry);
Message<String> testMessage1 = MessageBuilder.fromPayload("test")
.setReturnAddress(replyChannel1).build();
@@ -144,7 +144,7 @@ public class HandlerEndpointTests {
return null;
}
};
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setOutputChannelName("replyChannel");
endpoint.send(new StringMessage("test"));
@@ -156,7 +156,7 @@ public class HandlerEndpointTests {
@Test(expected=MessageRejectedException.class)
public void testEndpointWithSelectorRejecting() {
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.nullHandler());
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(TestHandlers.nullHandler());
endpoint.setSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
return false;
@@ -168,7 +168,7 @@ public class HandlerEndpointTests {
@Test
public void testEndpointWithSelectorAccepting() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countDownHandler(latch));
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(TestHandlers.countDownHandler(latch));
endpoint.setSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
return true;
@@ -182,7 +182,7 @@ public class HandlerEndpointTests {
@Test
public void testEndpointWithMultipleSelectorsAndFirstRejects() {
final AtomicInteger counter = new AtomicInteger();
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter));
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(TestHandlers.countingHandler(counter));
MessageSelectorChain selectorChain = new MessageSelectorChain();
selectorChain.add(new MessageSelector() {
public boolean accept(Message<?> message) {
@@ -212,7 +212,7 @@ public class HandlerEndpointTests {
public void testEndpointWithMultipleSelectorsAndFirstAccepts() {
final AtomicInteger selectorCounter = new AtomicInteger();
AtomicInteger handlerCounter = new AtomicInteger();
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(handlerCounter));
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(TestHandlers.countingHandler(handlerCounter));
MessageSelectorChain selectorChain = new MessageSelectorChain();
selectorChain.add(new MessageSelector() {
public boolean accept(Message<?> message) {
@@ -242,7 +242,7 @@ public class HandlerEndpointTests {
@Test
public void testEndpointWithMultipleSelectorsAndBothAccept() {
final AtomicInteger counter = new AtomicInteger();
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter));
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(TestHandlers.countingHandler(counter));
MessageSelectorChain selectorChain = new MessageSelectorChain();
selectorChain.add(new MessageSelector() {
public boolean accept(Message<?> message) {
@@ -264,7 +264,7 @@ public class HandlerEndpointTests {
@Test
public void testCorrelationId() {
QueueChannel replyChannel = new QueueChannel(1);
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return message;
}
@@ -279,7 +279,7 @@ public class HandlerEndpointTests {
@Test
public void testCorrelationIdSetByHandlerTakesPrecedence() {
QueueChannel replyChannel = new QueueChannel(1);
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return MessageBuilder.fromMessage(message)
.setCorrelationId("ABC-123").build();