From 4a13e38fde0bf1f90e4cfec81c8bc4ad153e58e0 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 2 Dec 2009 02:37:51 +0000 Subject: [PATCH] INT-901 removed ReplyMessageHolder and refactored AbstractReplyProducingMessageHandler accordingly --- .../AbstractRemotingOutboundGateway.java | 10 +-- .../file/FileWritingMessageHandler.java | 12 ++-- .../http/HttpOutboundEndpoint.java | 5 +- .../integration/jms/JmsOutboundGateway.java | 8 +-- .../ws/AbstractWebServiceOutboundGateway.java | 6 +- .../integration/filter/MessageFilter.java | 24 +++---- .../AbstractReplyProducingMessageHandler.java | 43 +++++++++--- .../integration/handler/BridgeHandler.java | 4 +- .../handler/ReplyMessageHolder.java | 70 ------------------- .../handler/ServiceActivatingHandler.java | 7 +- .../splitter/AbstractMessageSplitter.java | 26 ++++--- .../MessageTransformingHandler.java | 8 +-- .../ApplicationContextMessageBusTests.java | 43 +++++++----- .../bus/DirectChannelSubscriptionTests.java | 5 +- .../channel/MessageChannelTemplateTests.java | 5 +- .../integration/config/ChainParserTests.java | 5 +- ...tractReplyProducingMessageHandlerTest.java | 4 +- 17 files changed, 118 insertions(+), 167 deletions(-) delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyMessageHolder.java diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/AbstractRemotingOutboundGateway.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/AbstractRemotingOutboundGateway.java index 4279903cb5..e3f188fcc5 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/AbstractRemotingOutboundGateway.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/AbstractRemotingOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -21,7 +21,6 @@ import java.io.Serializable; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandlingException; import org.springframework.remoting.RemoteAccessException; @@ -52,7 +51,7 @@ public abstract class AbstractRemotingOutboundGateway extends AbstractReplyProdu @Override - public final void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + public final Object handleRequestMessage(Message message) { if (!(message.getPayload() instanceof Serializable)) { throw new MessageHandlingException(message, this.getClass().getName() + " expects a Serializable payload type " + @@ -60,10 +59,7 @@ public abstract class AbstractRemotingOutboundGateway extends AbstractReplyProdu } Message requestMessage = MessageBuilder.fromMessage(message).build(); try { - Message reply = this.handlerProxy.handle(requestMessage); - if (reply != null) { - replyHolder.set(reply); - } + return this.handlerProxy.handle(requestMessage); } catch (RemoteAccessException e) { throw new MessageHandlingException(message, "unable to handle message remotely", e); diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index 13247f85fa..ba6bf2a5cb 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -29,7 +29,6 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.integration.core.Message; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; @@ -150,7 +149,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } @Override - protected void handleRequestMessage(Message requestMessage, ReplyMessageHolder replyMessageHolder) { + protected Object handleRequestMessage(Message requestMessage) { Assert.notNull(requestMessage, "message must not be null"); Object payload = requestMessage.getPayload(); Assert.notNull(payload, "message payload must not be null"); @@ -180,14 +179,11 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } if (resultFile != null) { if (originalFileFromHeader == null && payload instanceof File) { - replyMessageHolder.set(MessageBuilder.withPayload(resultFile) - .setHeader(FileHeaders.ORIGINAL_FILE, (File) payload) - .build()); - } - else { - replyMessageHolder.set(resultFile); + return MessageBuilder.withPayload(resultFile) + .setHeader(FileHeaders.ORIGINAL_FILE, (File) payload); } } + return resultFile; } /** diff --git a/org.springframework.integration.http/src/main/java/org/springframework/integration/http/HttpOutboundEndpoint.java b/org.springframework.integration.http/src/main/java/org/springframework/integration/http/HttpOutboundEndpoint.java index fbe4883baa..f5f9c74b4e 100644 --- a/org.springframework.integration.http/src/main/java/org/springframework/integration/http/HttpOutboundEndpoint.java +++ b/org.springframework.integration.http/src/main/java/org/springframework/integration/http/HttpOutboundEndpoint.java @@ -25,7 +25,6 @@ import java.net.URL; import org.springframework.integration.core.Message; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageHandlingException; import org.springframework.util.Assert; import org.springframework.util.FileCopyUtils; @@ -85,12 +84,12 @@ public class HttpOutboundEndpoint extends AbstractReplyProducingMessageHandler { } @Override - protected void handleRequestMessage(Message requestMessage, ReplyMessageHolder replyMessageHolder) { + protected Object handleRequestMessage(Message requestMessage) { try { HttpRequest request = this.requestMapper.fromMessage(requestMessage); HttpResponse response = this.requestExecutor.executeRequest(request); Object reply = this.createReplyFromResponse(response); - replyMessageHolder.set(reply); + return reply; } catch (Exception e) { throw new MessageHandlingException(requestMessage, "failed to execute HTTP request", e); diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 59d30ae8a2..d7f6839f99 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -33,7 +33,6 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageTimeoutException; @@ -285,7 +284,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp } @Override - protected void handleRequestMessage(final Message message, final ReplyMessageHolder replyMessageHolder) { + protected Object handleRequestMessage(final Message message) { if (!this.initialized) { this.afterPropertiesSet(); } @@ -296,8 +295,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp throw new MessageTimeoutException(message, "failed to receive JMS response within timeout of: " + this.receiveTimeout + "ms"); } - Object result = this.messageConverter.fromMessage(jmsReply); - replyMessageHolder.set(result); + return this.messageConverter.fromMessage(jmsReply); } catch (JMSException e) { throw new MessageHandlingException(requestMessage, e); diff --git a/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceOutboundGateway.java b/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceOutboundGateway.java index 05ad08db9f..6d2a4be842 100644 --- a/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceOutboundGateway.java +++ b/org.springframework.integration.ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceOutboundGateway.java @@ -21,7 +21,6 @@ import java.io.IOException; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.ws.WebServiceMessage; @@ -95,15 +94,16 @@ public abstract class AbstractWebServiceOutboundGateway extends AbstractReplyPro } @Override - public final void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + public final Object handleRequestMessage(Message message) { Object responsePayload = this.doHandle(message.getPayload(), this.getRequestCallback(message)); if (responsePayload != null) { boolean shouldIgnore = (this.ignoreEmptyResponses && responsePayload instanceof String && !StringUtils.hasText((String) responsePayload)); if (!shouldIgnore) { - replyHolder.set(responsePayload); + return responsePayload; } } + return null; } protected abstract Object doHandle(Object requestPayload, WebServiceMessageCallback requestCallback); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java b/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java index 241a250819..2c7f9c84a7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java @@ -19,7 +19,6 @@ package org.springframework.integration.filter; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.selector.MessageSelector; @@ -83,22 +82,21 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { } @Override - protected void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + protected Object handleRequestMessage(Message message) { if (this.selector.accept(message)) { - replyHolder.set(message); + return message; } - else { - if (this.discardChannel != null) { - boolean discarded = this.sendReplyMessage(message, this.discardChannel); - if (!discarded) { - throw new MessageDeliveryException(message, - "failed to send rejected Message to the discard channel"); - } - } - if (this.throwExceptionOnRejection) { - throw new MessageRejectedException(message); + if (this.discardChannel != null) { + boolean discarded = this.sendReplyMessage(message, this.discardChannel); + if (!discarded) { + throw new MessageDeliveryException(message, + "failed to send rejected Message to the discard channel"); } } + if (this.throwExceptionOnRejection) { + throw new MessageRejectedException(message); + } + return null; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index a0e5ffbec3..f0d4742165 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -96,10 +96,10 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa * {@inheritDoc} */ @Override + @SuppressWarnings("unchecked") protected final void handleMessageInternal(Message message) { - ReplyMessageHolder replyMessageHolder = new ReplyMessageHolder(); - this.handleRequestMessage(message, replyMessageHolder); - if (replyMessageHolder.isEmpty()) { + Object result = this.handleRequestMessage(message); + if (result == null) { if (this.requiresReply) { throw new MessageHandlingException(message, "handler '" + this + "' requires a reply, but no reply was received"); @@ -111,9 +111,11 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel, this.channelResolver); MessageHeaders requestHeaders = message.getHeaders(); - for (MessageBuilder builder : replyMessageHolder.builders()) { - builder.copyHeadersIfAbsent(requestHeaders); - Message replyMessage = builder.build(); + if (result instanceof Iterable) { + this.sendReplyMessages((Iterable) result, requestHeaders, replyChannel); + } + else { + Message replyMessage = this.createReplyMessage(result, requestHeaders); if (!this.sendReplyMessage(replyMessage, replyChannel)) { throw new MessageDeliveryException(replyMessage, "failed to send reply Message to channel '" + replyChannel + "'"); @@ -121,7 +123,24 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } } - protected abstract void handleRequestMessage(Message requestMessage, ReplyMessageHolder replyMessageHolder); + private void sendReplyMessages(Iterable replies, MessageHeaders requestHeaders, MessageChannel replyChannel) { + for (Object reply : replies) { + Message replyMessage = this.createReplyMessage(reply, requestHeaders); + if (!this.sendReplyMessage(replyMessage, replyChannel)) { + throw new MessageDeliveryException(replyMessage, + "failed to send reply Message to channel '" + replyChannel + "'"); + } + } + } + + @SuppressWarnings("unchecked") + private Message createReplyMessage(Object reply, MessageHeaders requestHeaders) { + MessageBuilder builder = (reply instanceof MessageBuilder) ? (MessageBuilder) reply + : (reply instanceof Message) ? MessageBuilder.fromMessage((Message) reply) + : MessageBuilder.withPayload(reply); + builder.copyHeadersIfAbsent(requestHeaders); + return builder.build(); + } protected boolean sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { if (logger.isDebugEnabled()) { @@ -130,4 +149,12 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa return this.channelTemplate.send(replyMessage, replyChannel); } + /** + * Subclasses must implement this method to handle the request Message. The return + * value may be a Message, a MessageBuilder, or any plain Object. The base class + * will handle the final creation of a reply Message from any of those starting + * points. If the return value is null, the Message flow will end here. + */ + protected abstract Object handleRequestMessage(Message requestMessage); + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/BridgeHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/BridgeHandler.java index 1b0e01ceb0..3c722943ee 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/BridgeHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/BridgeHandler.java @@ -36,11 +36,11 @@ import org.springframework.util.Assert; public class BridgeHandler extends AbstractReplyProducingMessageHandler { @Override - protected void handleRequestMessage(Message requestMessage, ReplyMessageHolder replyMessageHolder) { + protected Object handleRequestMessage(Message requestMessage) { if (requestMessage.getHeaders().getReplyChannel() == null) { this.verifyOutputChannel(); } - replyMessageHolder.set(requestMessage); + return requestMessage; } private void verifyOutputChannel() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyMessageHolder.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyMessageHolder.java deleted file mode 100644 index db86c998de..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyMessageHolder.java +++ /dev/null @@ -1,70 +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 java.util.ArrayList; -import java.util.Collections; -import java.util.List; - -import org.springframework.integration.core.Message; -import org.springframework.integration.message.MessageBuilder; - -/** - * @author Mark Fisher - */ -public class ReplyMessageHolder { - - private final List> builders = new ArrayList>(); - - - public MessageBuilder set(Object messageOrPayload) { - return this.createAndAddBuilder(messageOrPayload, true); - } - - public MessageBuilder add(Object messageOrPayload) { - return this.createAndAddBuilder(messageOrPayload, false); - } - - public boolean isEmpty() { - return this.builders.isEmpty(); - } - - public List> builders() { - return Collections.unmodifiableList(this.builders); - } - - private MessageBuilder createAndAddBuilder(Object messageOrPayload, boolean clearExistingValues) { - MessageBuilder builder = null; - if (messageOrPayload instanceof MessageBuilder) { - builder = (MessageBuilder) messageOrPayload; - } - else if (messageOrPayload instanceof Message) { - builder = MessageBuilder.fromMessage((Message) messageOrPayload); - } - else { - builder = MessageBuilder.withPayload(messageOrPayload); - } - synchronized (this.builders) { - if (clearExistingValues) { - this.builders.clear(); - } - this.builders.add(builder); - } - return builder; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java index aab167539b..c623a0426d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java @@ -44,12 +44,9 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl @Override - protected void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + protected Object handleRequestMessage(Message message) { try { - Object result = this.processor.processMessage(message); - if (result != null) { - replyHolder.set(result); - } + return this.processor.processMessage(message); } catch (Exception e) { if (e instanceof RuntimeException) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index abfba6bd97..220df43eed 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -16,13 +16,15 @@ package org.springframework.integration.splitter; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import java.util.UUID; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; +import org.springframework.integration.message.MessageBuilder; /** * Base class for Message-splitting handlers. @@ -32,19 +34,20 @@ import org.springframework.integration.handler.ReplyMessageHolder; public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler { @Override - protected final void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + protected final Object handleRequestMessage(Message message) { Object result = this.splitMessage(message); if (result == null) { - return; + return null; } Object correlationId = (message.getHeaders().getCorrelationId() != null) ? - message.getHeaders().getCorrelationId(): message.getHeaders().getId(); + message.getHeaders().getCorrelationId() : message.getHeaders().getId(); + List> messageBuilders = new ArrayList>(); if (result instanceof Collection) { Collection items = (Collection) result; int sequenceNumber = 0; int sequenceSize = items.size(); for (Object item : items) { - this.addReply(replyHolder, item, correlationId, ++sequenceNumber, sequenceSize); + messageBuilders.add(this.createBuilder(item, correlationId, ++sequenceNumber, sequenceSize)); } } else if (result.getClass().isArray()) { @@ -52,19 +55,24 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess int sequenceNumber = 0; int sequenceSize = items.length; for (Object item : items) { - this.addReply(replyHolder, item, correlationId, ++sequenceNumber, sequenceSize); + messageBuilders.add(this.createBuilder(item, correlationId, ++sequenceNumber, sequenceSize)); } } else { - this.addReply(replyHolder, result, correlationId, 1, 1); + messageBuilders.add(this.createBuilder(result, correlationId, 1, 1)); } + return messageBuilders; } - private void addReply(ReplyMessageHolder replyHolder, Object item, Object correlationId, int sequenceNumber, int sequenceSize) { - replyHolder.add(item).setCorrelationId(correlationId) + @SuppressWarnings("unchecked") + private MessageBuilder createBuilder(Object item, Object correlationId, int sequenceNumber, int sequenceSize) { + MessageBuilder builder = (item instanceof Message) ? + MessageBuilder.fromMessage((Message) item) : MessageBuilder.withPayload(item); + builder.setCorrelationId(correlationId) .setSequenceNumber(sequenceNumber) .setSequenceSize(sequenceSize) .setHeader(MessageHeaders.ID, UUID.randomUUID()); + return builder; } /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java index 42b5d2ea84..5165cccc3d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java @@ -18,7 +18,6 @@ package org.springframework.integration.transformer; import org.springframework.integration.core.Message; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.util.Assert; /** @@ -44,12 +43,9 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan @Override - protected void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + protected Object handleRequestMessage(Message message) { try { - Message result = transformer.transform(message); - if (result != null) { - replyHolder.set(result); - } + return transformer.transform(message); } catch (Exception e) { if (e instanceof MessageTransformationException) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index 951fab0633..45c53dbe05 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java @@ -16,8 +16,16 @@ package org.springframework.integration.bus; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +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 org.junit.Test; + import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.PublishSubscribeChannel; @@ -28,15 +36,15 @@ import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; -import org.springframework.integration.message.*; +import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.StringMessage; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.test.util.TestUtils.TestApplicationContext; import org.springframework.scheduling.support.PeriodicTrigger; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - /** * @author Mark Fisher */ @@ -53,8 +61,8 @@ public class ApplicationContextMessageBusTests { .setReplyChannelName("targetChannel").build(); sourceChannel.send(message); AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { - replyHolder.set(message); + public Object handleRequestMessage(Message message) { + return message; } }; handler.setBeanFactory(context); @@ -99,14 +107,14 @@ public class ApplicationContextMessageBusTests { QueueChannel outputChannel2 = new QueueChannel(); AbstractReplyProducingMessageHandler handler1 = new AbstractReplyProducingMessageHandler() { @Override - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { - replyHolder.set(message); + public Object handleRequestMessage(Message message) { + return message; } }; AbstractReplyProducingMessageHandler handler2 = new AbstractReplyProducingMessageHandler() { @Override - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { - replyHolder.set(message); + public Object handleRequestMessage(Message message) { + return message; } }; context.registerChannel("input", inputChannel); @@ -135,16 +143,16 @@ public class ApplicationContextMessageBusTests { final CountDownLatch latch = new CountDownLatch(2); AbstractReplyProducingMessageHandler handler1 = new AbstractReplyProducingMessageHandler() { @Override - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { - replyHolder.set(message); + public Object handleRequestMessage(Message message) { latch.countDown(); + return message; } }; AbstractReplyProducingMessageHandler handler2 = new AbstractReplyProducingMessageHandler() { @Override - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { - replyHolder.set(message); + public Object handleRequestMessage(Message message) { latch.countDown(); + return message; } }; context.registerChannel("input", inputChannel); @@ -198,8 +206,9 @@ public class ApplicationContextMessageBusTests { final CountDownLatch latch = new CountDownLatch(1); AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @Override - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + public Object handleRequestMessage(Message message) { latch.countDown(); + return null; } }; PollingConsumer endpoint = new PollingConsumer(errorChannel, handler); 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 0a6e935956..7adc972568 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 @@ -32,7 +32,6 @@ import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.message.StringMessage; import org.springframework.integration.test.util.TestUtils; @@ -89,7 +88,7 @@ public class DirectChannelSubscriptionTests { public void exceptionThrownFromRegisteredEndpoint() { AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @Override - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { + public Object handleRequestMessage(Message message) { throw new RuntimeException("intentional test failure"); } }; @@ -124,7 +123,7 @@ public class DirectChannelSubscriptionTests { } - private static class TestBean { + static class TestBean { public Message handle(Message message) { return new StringMessage(message.getPayload() + "!"); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java index 6f3d75d677..7a18e74673 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java @@ -34,7 +34,6 @@ import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.StringMessage; import org.springframework.integration.test.util.TestUtils; @@ -57,8 +56,8 @@ public class MessageChannelTemplateTests { context.registerChannel("requestChannel", requestChannel); AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @Override - public void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { - replyHolder.set(message.getPayload().toString().toUpperCase()); + public Object handleRequestMessage(Message message) { + return message.getPayload().toString().toUpperCase(); } }; PollingConsumer endpoint = new PollingConsumer(requestChannel, handler); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java index 16e4ca00c4..490bc891b1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java @@ -31,7 +31,6 @@ import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -149,8 +148,8 @@ public class ChainParserTests { public static class StubHandler extends AbstractReplyProducingMessageHandler { @Override - protected void handleRequestMessage(Message requestMessage, ReplyMessageHolder replyMessageHolder) { - replyMessageHolder.add(successMessage); + protected Object handleRequestMessage(Message requestMessage) { + return successMessage; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTest.java index 1ceaf2d060..f4a5929a6b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTest.java @@ -35,8 +35,8 @@ public class AbstractReplyProducingMessageHandlerTest { private AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @Override - protected void handleRequestMessage(Message requestMessage, ReplyMessageHolder replyMessageHolder) { - replyMessageHolder.add(requestMessage); + protected Object handleRequestMessage(Message requestMessage) { + return requestMessage; } }; private Message message = MessageBuilder.withPayload("test").build();