diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java index ff3a0b1589..bf870b9bf7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java @@ -24,6 +24,7 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.message.CompositeMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageHeaders; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.message.MessageTarget; @@ -37,6 +38,8 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint { private volatile MessageSelector selector; + private volatile boolean requiresReply = false; + private final List interceptors = new CopyOnWriteArrayList(); @@ -44,6 +47,10 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint { this.selector = selector; } + public void setRequiresReply(boolean requiresReply) { + this.requiresReply = requiresReply; + } + public void addInterceptor(EndpointInterceptor interceptor) { this.interceptors.add(interceptor); } @@ -68,15 +75,21 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint { } Object result = this.handle(message); if (result == null) { - return false; + if (this.requiresReply) { + throw new MessageHandlingException(message, "endpoint '" + this.getName() + + " requires a reply, but no reply was received"); + } + return true; } Message reply = buildReplyMessage(result, message.getHeaders()); MessageTarget replyTarget = this.resolveReplyTarget(message); if (reply instanceof CompositeMessage && this.shouldSplitComposite()) { + boolean sentAtLeastOne = false; for (Message nextReply : (CompositeMessage) reply) { - this.sendReplyMessage(nextReply, replyTarget); + boolean sent = this.sendReplyMessage(nextReply, replyTarget); + sentAtLeastOne = (sentAtLeastOne || sent); } - return true; + return sentAtLeastOne; } else { return this.sendReplyMessage(reply, replyTarget); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java deleted file mode 100644 index b065852e57..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java +++ /dev/null @@ -1,205 +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 java.lang.reflect.Array; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; - -import org.springframework.integration.channel.ChannelRegistry; -import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.CompositeMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.message.MessageHandlingException; -import org.springframework.integration.message.MessageHeaders; -import org.springframework.integration.message.MessageRejectedException; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.util.Assert; - -/** - * The default Message Endpoint implementation. Serves as a "host" for any - * {@link MessageHandler} and resolves the target for any reply Message(s) - * returned by that handler. If the handler returns a non-empty - * {@link CompositeMessage}, each Message in its list will be sent - * as a separate reply Message. - * - *

The reply target is resolved according to the following order: - *

    - *
  1. the 'outputChannel' of this Message Endpoint
    2. - *
  2. the 'returnAddress' header value of the request Message
    3. - *
- * For the 'returnAddress' value, either a {@link MessageTarget} instance - * or String is accepted. If the value is a String, then the endpoint will - * consult its {@link ChannelRegistry} (typically provided by the MessageBus). - * If no reply target can be determined for a non-null reply Message, - * a {@link MessageEndpointReplyException} will be thrown. - * - * @author Mark Fisher - */ -public class DefaultEndpoint extends AbstractRequestReplyEndpoint { - - private final T handler; - - private volatile MessageSelector selector; - - private final List interceptors = new ArrayList(); - - - /** - * Create an endpoint for the given handler. - */ - public DefaultEndpoint(T handler) { - Assert.notNull(handler, "handler must not be null"); - this.handler = handler; - } - - - protected T getHandler() { - return this.handler; - } - - @Override - public void setChannelRegistry(ChannelRegistry channelRegistry) { - super.setChannelRegistry(channelRegistry); - if (this.handler instanceof ChannelRegistryAware) { - ((ChannelRegistryAware) this.handler).setChannelRegistry(channelRegistry); - } - } - - public void setSelector(MessageSelector selector) { - this.selector = selector; - } - - public void addInterceptor(EndpointInterceptor interceptor) { - this.interceptors.add(interceptor); - } - - public void setInterceptors(List interceptors) { - this.interceptors.clear(); - for (EndpointInterceptor interceptor : interceptors) { - this.addInterceptor(interceptor); - } - } - - /** - * Specify the timeout for sending reply Messages to the reply - * target. The default value indicates an indefinite timeout. - */ - public void setReplyTimeout(long replyTimeout) { - this.getMessageExchangeTemplate().setSendTimeout(replyTimeout); - } - - @Override - public final Message handleRequestMessage(Message requestMessage) { - if (requestMessage == null || requestMessage.getPayload() == null) { - return null; - } - for (EndpointInterceptor interceptor : this.interceptors) { - requestMessage = interceptor.preHandle(requestMessage); - if (requestMessage == null) { - return null; - } - } - if (!this.supports(requestMessage)) { - throw new MessageRejectedException(requestMessage, "unsupported message"); - } - Message replyMessage = this.handler.handle(requestMessage); - for (int i = this.interceptors.size() - 1; i >= 0; i--) { - EndpointInterceptor interceptor = this.interceptors.get(i); - if (interceptor != null) { - replyMessage = interceptor.postHandle(replyMessage); - } - } - return replyMessage; - } - - /** - * The reply Message is considered invalid if it is null, has a null payload, - * or has an empty Array or Collection as its payload. - */ - @Override - protected boolean isValidReplyMessage(Message replyMessage) { - if (replyMessage == null) { - return false; - } - Object payload = replyMessage.getPayload(); - if (payload == null) { - return false; - } - if (payload.getClass().isArray() && Array.getLength(payload) == 0) { - return false; - } - if (payload instanceof Collection && ((Collection) payload).size() == 0) { - return false; - } - return true; - } - - @Override - protected void sendReplyMessage(Message replyMessage, Message requestMessage) { - if (replyMessage == null) { - throw new MessageHandlingException(requestMessage, "reply message must not be null"); - } - MessageTarget replyTarget = this.resolveReplyTarget(replyMessage, requestMessage.getHeaders()); - if (replyTarget == null) { - throw new MessageEndpointReplyException(replyMessage, requestMessage, - "unable to resolve reply target"); - } - replyMessage = MessageBuilder.fromMessage(replyMessage) - .copyHeadersIfAbsent(requestMessage.getHeaders()) - .setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestMessage.getHeaders().getId()) - .build(); - if (!this.getMessageExchangeTemplate().send(replyMessage, replyTarget)) { - throw new MessageEndpointReplyException(replyMessage, requestMessage, - "failed to send reply to '" + replyTarget + "'"); - } - } - - protected boolean supports(Message message) { - if (this.selector != null && !this.selector.accept(message)) { - if (logger.isDebugEnabled()) { - logger.debug("selector for endpoint '" + this + "' rejected message: " + message); - } - return false; - } - return true; - } - - private MessageTarget resolveReplyTarget(Message replyMessage, MessageHeaders requestHeaders) { - MessageTarget replyTarget = this.getTarget(); - if (replyTarget == null) { - Object returnAddress = requestHeaders.getReturnAddress(); - if (returnAddress != null) { - if (returnAddress instanceof MessageTarget) { - replyTarget = (MessageTarget) returnAddress; - } - else if (returnAddress instanceof String) { - ChannelRegistry registry = this.getChannelRegistry(); - if (registry != null) { - replyTarget = registry.lookupChannel((String) returnAddress); - } - } - } - } - return replyTarget; - } - -} 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 fc4a2c03a9..d91ed9eb0e 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 @@ -32,9 +32,8 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.DefaultEndpoint; +import org.springframework.integration.endpoint.AbstractInOutEndpoint; import org.springframework.integration.endpoint.InboundChannelAdapter; -import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -60,12 +59,11 @@ public class DefaultMessageBusTests { .setReturnAddress("targetChannel").build(); sourceChannel.send(message); bus.registerChannel(targetChannel); - MessageHandler handler = new MessageHandler() { + AbstractInOutEndpoint endpoint = new AbstractInOutEndpoint() { public Message handle(Message message) { return message; } }; - DefaultEndpoint endpoint = new DefaultEndpoint(handler); endpoint.setBeanName("testEndpoint"); endpoint.setSource(sourceChannel); bus.registerEndpoint(endpoint); @@ -109,12 +107,12 @@ public class DefaultMessageBusTests { QueueChannel inputChannel = new QueueChannel(); QueueChannel outputChannel1 = new QueueChannel(); QueueChannel outputChannel2 = new QueueChannel(); - MessageHandler handler1 = new MessageHandler() { + AbstractInOutEndpoint endpoint1 = new AbstractInOutEndpoint() { public Message handle(Message message) { return MessageBuilder.fromMessage(message).build(); } }; - MessageHandler handler2 = new MessageHandler() { + AbstractInOutEndpoint endpoint2 = new AbstractInOutEndpoint() { public Message handle(Message message) { return MessageBuilder.fromMessage(message).build(); } @@ -126,11 +124,9 @@ public class DefaultMessageBusTests { bus.registerChannel(inputChannel); bus.registerChannel(outputChannel1); bus.registerChannel(outputChannel2); - DefaultEndpoint endpoint1 = new DefaultEndpoint(handler1); endpoint1.setBeanName("testEndpoint1"); endpoint1.setSource(inputChannel); endpoint1.setTarget(outputChannel1); - DefaultEndpoint endpoint2 = new DefaultEndpoint(handler2); endpoint2.setBeanName("testEndpoint2"); endpoint2.setSource(inputChannel); endpoint2.setTarget(outputChannel2); @@ -150,14 +146,14 @@ public class DefaultMessageBusTests { QueueChannel outputChannel1 = new QueueChannel(); QueueChannel outputChannel2 = new QueueChannel(); final CountDownLatch latch = new CountDownLatch(2); - MessageHandler handler1 = new MessageHandler() { + AbstractInOutEndpoint endpoint1 = new AbstractInOutEndpoint() { public Message handle(Message message) { Message reply = MessageBuilder.fromMessage(message).build(); latch.countDown(); return reply; } }; - MessageHandler handler2 = new MessageHandler() { + AbstractInOutEndpoint endpoint2 = new AbstractInOutEndpoint() { public Message handle(Message message) { Message reply = MessageBuilder.fromMessage(message).build(); latch.countDown(); @@ -171,11 +167,9 @@ public class DefaultMessageBusTests { bus.registerChannel(inputChannel); bus.registerChannel(outputChannel1); bus.registerChannel(outputChannel2); - DefaultEndpoint endpoint1 = new DefaultEndpoint(handler1); endpoint1.setBeanName("testEndpoint1"); endpoint1.setSource(inputChannel); endpoint1.setTarget(outputChannel1); - DefaultEndpoint endpoint2 = new DefaultEndpoint(handler2); endpoint2.setBeanName("testEndpoint2"); endpoint2.setSource(inputChannel); endpoint2.setTarget(outputChannel2); @@ -232,13 +226,12 @@ public class DefaultMessageBusTests { errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME); bus.registerChannel(errorChannel); final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler = new MessageHandler() { + AbstractInOutEndpoint endpoint = new AbstractInOutEndpoint() { public Message handle(Message message) { latch.countDown(); return null; } }; - DefaultEndpoint endpoint = new DefaultEndpoint(handler); endpoint.setBeanName("testEndpoint"); endpoint.setSource(errorChannel); bus.registerEndpoint(endpoint); 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 b8c8161eb8..f48b3747d0 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,8 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor; -import org.springframework.integration.endpoint.DefaultEndpoint; +import org.springframework.integration.endpoint.AbstractInOutEndpoint; +import org.springframework.integration.endpoint.ServiceActivatorEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; @@ -57,7 +58,7 @@ public class DirectChannelSubscriptionTests { @Test public void testSendAndReceiveForRegisteredEndpoint() { - DefaultEndpoint endpoint = new DefaultEndpoint(new TestHandler()); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new TestHandler()); endpoint.setSource(sourceChannel); endpoint.setTarget(targetChannel); endpoint.setBeanName("testEndpoint"); @@ -87,11 +88,11 @@ public class DirectChannelSubscriptionTests { QueueChannel errorChannel = new QueueChannel(); errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME); bus.registerChannel(errorChannel); - DefaultEndpoint endpoint = new DefaultEndpoint(new MessageHandler() { + AbstractInOutEndpoint endpoint = new AbstractInOutEndpoint() { public Message handle(Message message) { throw new RuntimeException("intentional test failure"); } - }); + }; endpoint.setSource(sourceChannel); endpoint.setTarget(targetChannel); endpoint.setBeanName("testEndpoint"); 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 c10917e2ca..5c572c7973 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 @@ - + @@ -21,7 +21,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 ce9e00845a..5758e4969c 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 @@ -26,7 +26,8 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; -import org.springframework.integration.endpoint.DefaultEndpoint; +import org.springframework.integration.endpoint.AbstractInOutEndpoint; +import org.springframework.integration.endpoint.ServiceActivatorEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.Message; @@ -131,9 +132,9 @@ public class SimpleDispatcherTests { final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger counter3 = new AtomicInteger(); final AtomicInteger selectorCounter = new AtomicInteger(); - DefaultEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)); - DefaultEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)); - DefaultEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch)); + AbstractInOutEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)); + AbstractInOutEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)); + AbstractInOutEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch)); endpoint1.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint2.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint3.setSelector(new TestMessageSelector(selectorCounter, true)); @@ -156,9 +157,9 @@ public class SimpleDispatcherTests { final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger counter3 = new AtomicInteger(); final AtomicInteger selectorCounter = new AtomicInteger(); - DefaultEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)); - DefaultEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)); - DefaultEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch)); + AbstractInOutEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)); + AbstractInOutEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)); + AbstractInOutEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch)); endpoint1.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint2.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint3.setSelector(new TestMessageSelector(selectorCounter, false)); @@ -222,8 +223,8 @@ public class SimpleDispatcherTests { } - private static DefaultEndpoint createEndpoint(MessageHandler handler) { - return new DefaultEndpoint(handler); + private static ServiceActivatorEndpoint createEndpoint(MessageHandler handler) { + return new ServiceActivatorEndpoint(handler); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java similarity index 86% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java index 55c2a9c0d5..bab6761605 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java @@ -16,16 +16,16 @@ package org.springframework.integration.endpoint; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + import org.junit.Test; import org.springframework.integration.bus.DefaultMessageBus; @@ -38,6 +38,7 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageRejectedException; +import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.MessageSelector; import org.springframework.integration.message.selector.MessageSelectorChain; @@ -46,13 +47,13 @@ import org.springframework.integration.message.selector.MessageSelectorChain; * @author Mark Fisher * @author Marius Bogoevici */ -public class DefaultEndpointTests { +public class ServiceActivatorEndpointTests { @Test public void outputChannel() { QueueChannel channel = new QueueChannel(1); MessageHandler handler = new TestHandler(); - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); endpoint.setTarget(channel); Message message = MessageBuilder.fromPayload("foo").build(); endpoint.send(message); @@ -65,7 +66,7 @@ public class DefaultEndpointTests { public void outputChannelTakesPrecedence() { QueueChannel channel1 = new QueueChannel(1); QueueChannel channel2 = new QueueChannel(1); - DefaultEndpoint endpoint = new DefaultEndpoint(new TestHandler()); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new TestHandler()); endpoint.setTarget(channel1); Message message = MessageBuilder.fromPayload("foo").setReturnAddress(channel2).build(); endpoint.send(message); @@ -80,7 +81,7 @@ public class DefaultEndpointTests { public void returnAddressHeader() { QueueChannel channel = new QueueChannel(1); MessageHandler handler = new TestHandler(); - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); Message message = MessageBuilder.fromPayload("foo").setReturnAddress(channel).build(); endpoint.send(message); Message reply = channel.receive(0); @@ -95,7 +96,7 @@ public class DefaultEndpointTests { ChannelRegistry channelRegistry = new DefaultMessageBus(); channelRegistry.registerChannel(channel); MessageHandler handler = new TestHandler(); - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); endpoint.setChannelRegistry(channelRegistry); Message message = MessageBuilder.fromPayload("foo").setReturnAddress("testChannel").build(); endpoint.send(message); @@ -116,7 +117,7 @@ public class DefaultEndpointTests { return new StringMessage("foo" + message.getPayload()); } }; - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); endpoint.setChannelRegistry(channelRegistry); Message testMessage1 = MessageBuilder.fromPayload("bar") .setReturnAddress(replyChannel1).build(); @@ -140,7 +141,7 @@ public class DefaultEndpointTests { public void noOutputChannelFallsBackToReturnAddress() { QueueChannel channel = new QueueChannel(1); MessageHandler handler = new TestHandler(); - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); Message message = MessageBuilder.fromPayload("foo").setReturnAddress(channel).build(); endpoint.send(message); Message reply = channel.receive(0); @@ -148,10 +149,10 @@ public class DefaultEndpointTests { assertEquals("FOO", reply.getPayload()); } - @Test(expected = MessageEndpointReplyException.class) + @Test(expected = MessagingException.class) public void noReplyTarget() { MessageHandler handler = new TestHandler(); - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); Message message = MessageBuilder.fromPayload("foo").build(); endpoint.send(message); } @@ -160,7 +161,7 @@ public class DefaultEndpointTests { public void noReplyMessage() { QueueChannel channel = new QueueChannel(1); MessageHandler handler = new TestNullReplyHandler(); - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); endpoint.setTarget(channel); Message message = MessageBuilder.fromPayload("foo").build(); endpoint.send(message); @@ -171,7 +172,7 @@ public class DefaultEndpointTests { public void noReplyMessageWithRequiresReply() { QueueChannel channel = new QueueChannel(1); MessageHandler handler = new TestNullReplyHandler(); - DefaultEndpoint endpoint = new DefaultEndpoint(handler); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler); endpoint.setRequiresReply(true); endpoint.setTarget(channel); Message message = MessageBuilder.fromPayload("foo").build(); @@ -180,7 +181,7 @@ public class DefaultEndpointTests { @Test(expected=MessageRejectedException.class) public void endpointWithSelectorRejecting() { - DefaultEndpoint endpoint = new DefaultEndpoint(TestHandlers.nullHandler()); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(TestHandlers.nullHandler()); endpoint.setSelector(new MessageSelector() { public boolean accept(Message message) { return false; @@ -192,7 +193,7 @@ public class DefaultEndpointTests { @Test public void endpointWithSelectorAccepting() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - DefaultEndpoint endpoint = new DefaultEndpoint(TestHandlers.countDownHandler(latch)); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(TestHandlers.countDownHandler(latch)); endpoint.setSelector(new MessageSelector() { public boolean accept(Message message) { return true; @@ -206,7 +207,7 @@ public class DefaultEndpointTests { @Test public void endpointWithMultipleSelectorsAndFirstRejects() { final AtomicInteger counter = new AtomicInteger(); - DefaultEndpoint endpoint = new DefaultEndpoint(TestHandlers.countingHandler(counter)); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(TestHandlers.countingHandler(counter)); MessageSelectorChain selectorChain = new MessageSelectorChain(); selectorChain.add(new MessageSelector() { public boolean accept(Message message) { @@ -236,7 +237,7 @@ public class DefaultEndpointTests { public void endpointWithMultipleSelectorsAndFirstAccepts() { final AtomicInteger selectorCounter = new AtomicInteger(); AtomicInteger handlerCounter = new AtomicInteger(); - DefaultEndpoint endpoint = new DefaultEndpoint(TestHandlers.countingHandler(handlerCounter)); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(TestHandlers.countingHandler(handlerCounter)); MessageSelectorChain selectorChain = new MessageSelectorChain(); selectorChain.add(new MessageSelector() { public boolean accept(Message message) { @@ -266,7 +267,7 @@ public class DefaultEndpointTests { @Test public void endpointWithMultipleSelectorsAndBothAccept() { final AtomicInteger counter = new AtomicInteger(); - DefaultEndpoint endpoint = new DefaultEndpoint(TestHandlers.countingHandler(counter)); + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(TestHandlers.countingHandler(counter)); MessageSelectorChain selectorChain = new MessageSelectorChain(); selectorChain.add(new MessageSelector() { public boolean accept(Message message) { @@ -288,7 +289,7 @@ public class DefaultEndpointTests { @Test public void correlationId() { QueueChannel replyChannel = new QueueChannel(1); - DefaultEndpoint endpoint = new DefaultEndpoint(new MessageHandler() { + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new MessageHandler() { public Message handle(Message message) { return message; } @@ -303,7 +304,7 @@ public class DefaultEndpointTests { @Test public void correlationIdSetByHandlerTakesPrecedence() { QueueChannel replyChannel = new QueueChannel(1); - DefaultEndpoint endpoint = new DefaultEndpoint(new MessageHandler() { + ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new MessageHandler() { public Message handle(Message message) { return MessageBuilder.fromMessage(message) .setCorrelationId("ABC-123").build(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageExchangeTemplateTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageExchangeTemplateTests.java index 5e3bba11ec..596ec48f78 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageExchangeTemplateTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageExchangeTemplateTests.java @@ -30,8 +30,7 @@ import org.junit.Test; import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.DefaultEndpoint; -import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.endpoint.AbstractInOutEndpoint; /** * @author Mark Fisher @@ -45,14 +44,13 @@ public class MessageExchangeTemplateTests { public void setUp() { this.requestChannel = new QueueChannel(); this.requestChannel.setBeanName("requestChannel"); - MessageHandler testHandler = new MessageHandler() { + AbstractInOutEndpoint endpoint = new AbstractInOutEndpoint() { public Message handle(Message message) { return new StringMessage(message.getPayload().toString().toUpperCase()); } }; MessageBus bus = new DefaultMessageBus(); bus.registerChannel(requestChannel); - DefaultEndpoint endpoint = new DefaultEndpoint(testHandler); endpoint.setBeanName("testEndpoint"); endpoint.setSource(requestChannel); bus.registerEndpoint(endpoint);