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 deleted file mode 100644 index 5f3f8a8c70..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java +++ /dev/null @@ -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 replyTimeout 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"); - } - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java index ddd513cfe7..755eb681f0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java @@ -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 endpoint = new SimpleEndpoint(handler); + endpoint.setBeanName("testEndpoint"); endpoint.setSource(channelAdapter); bus.registerEndpoint(endpoint); bus.start(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index d190e03e36..9b2ebc9125 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(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")); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml index cf33ea8556..fe46980d98 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -10,7 +10,7 @@ - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java index 852019b495..82873b99fa 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java @@ -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(handler1); + SimpleEndpoint endpoint2 = new SimpleEndpoint(handler2); + SimpleEndpoint endpoint3 = new SimpleEndpoint(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 createEndpoint(MessageHandler handler) { + return new SimpleEndpoint(handler); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java index 10e96a11d9..bf6f37da75 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java @@ -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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(handler); Message 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 endpoint = new SimpleEndpoint(handler); endpoint.setChannelRegistry(channelRegistry); Message 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 endpoint = new SimpleEndpoint(handler); endpoint.setChannelRegistry(channelRegistry); Message testMessage1 = MessageBuilder.fromPayload("test") .setReturnAddress(replyChannel1).build(); @@ -144,7 +144,7 @@ public class HandlerEndpointTests { return null; } }; - HandlerEndpoint endpoint = new HandlerEndpoint(handler); + SimpleEndpoint endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(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 endpoint = new SimpleEndpoint(new MessageHandler() { public Message handle(Message message) { return MessageBuilder.fromMessage(message) .setCorrelationId("ABC-123").build();