diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerGatewayTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerGatewayTests.java index 8f22e745d2..f168ec385c 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerGatewayTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerGatewayTests.java @@ -29,9 +29,9 @@ import java.util.concurrent.Executors; import org.junit.Test; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.StringMessage; import org.springframework.mock.web.MockHttpServletRequest; import org.springframework.mock.web.MockHttpServletResponse; @@ -64,8 +64,8 @@ public class HttpInvokerGatewayTests { Executors.newSingleThreadExecutor().execute(new Runnable() { public void run() { Message message = channel.receive(); - MessageTarget replyTarget = (MessageTarget) message.getHeaders().getReturnAddress(); - replyTarget.send(new StringMessage(message.getPayload().toString().toUpperCase())); + MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReturnAddress(); + replyChannel.send(new StringMessage(message.getPayload().toString().toUpperCase())); } }); HttpInvokerGateway gateway = new HttpInvokerGateway(channel); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java index 581d81833b..d7211fa66b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java @@ -30,13 +30,12 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.channel.BlockingChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.endpoint.AbstractInOutEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.BlockingTarget; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; -import org.springframework.integration.message.MessageTarget; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -199,19 +198,19 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoi private void afterRelease(Object correlationId, List> releasedMessages) { Message[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages); for (Message result : processedMessages) { - MessageTarget replyTarget = this.getTarget(); - if (replyTarget == null) { - replyTarget = this.resolveReplyTargetFromMessage(result); - if (replyTarget == null) { - replyTarget = this.resolveReplyTargetFromMessage(releasedMessages.get(0)); + MessageChannel replyChannel = this.getOutputChannel(); + if (replyChannel == null) { + replyChannel = this.resolveReplyChannelFromMessage(result); + if (replyChannel == null) { + replyChannel = this.resolveReplyChannelFromMessage(releasedMessages.get(0)); } } - if (replyTarget != null) { - if (replyTarget instanceof BlockingTarget && this.sendTimeout >= 0) { - ((BlockingTarget) replyTarget).send(result, this.sendTimeout); + if (replyChannel != null) { + if (replyChannel instanceof BlockingChannel && this.sendTimeout >= 0) { + ((BlockingChannel) replyChannel).send(result, this.sendTimeout); } else { - replyTarget.send(result); + replyChannel.send(result); } } else if (logger.isWarnEnabled()) { @@ -222,7 +221,10 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoi private void sendToDiscardChannelIfAvailable(Message message) { if (this.discardChannel != null) { - if (!this.discardChannel.send(message, this.sendTimeout)) { + boolean sent = (this.discardChannel instanceof BlockingChannel && this.sendTimeout >= 0) + ? ((BlockingChannel) this.discardChannel).send(message, this.sendTimeout) + : this.discardChannel.send(message); + if (!sent) { if (logger.isWarnEnabled()) { logger.warn("unable to send to 'discardChannel', message: " + message); } @@ -230,14 +232,14 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoi } } - protected MessageTarget resolveReplyTargetFromMessage(Message message) { + protected MessageChannel resolveReplyChannelFromMessage(Message message) { Object returnAddress = message.getHeaders().getReturnAddress(); if (returnAddress != null) { - if (returnAddress instanceof MessageTarget) { - return (MessageTarget) returnAddress; + if (returnAddress instanceof MessageChannel) { + return (MessageChannel) returnAddress; } if (logger.isWarnEnabled()) { - logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageTarget."); + logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageChannel."); } } return null; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 2ff947fab5..c2498d1e29 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -34,7 +34,7 @@ import org.springframework.integration.message.Message; * * @author Mark Fisher */ -public abstract class AbstractMessageChannel implements MessageChannel, BeanNameAware { +public abstract class AbstractMessageChannel implements BlockingChannel, BeanNameAware { private final Log logger = LogFactory.getLog(this.getClass()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java new file mode 100644 index 0000000000..e6259a6017 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java @@ -0,0 +1,52 @@ +/* + * 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.channel; + +import org.springframework.integration.message.Message; + +/** + * Extends the base MessageChannel interface for channels that may block when a + * Message is sent. Adds the timeout-aware {@link #send(Message, long)} method. + * + * @author Mark Fisher + */ +public interface BlockingChannel extends MessageChannel { + + /** + * Send a message, blocking indefinitely if necessary. + * + * @param message the {@link Message} to send + * + * @return true if the message is sent successfully, + * false if interrupted + */ + boolean send(Message message); + + /** + * Send a message, blocking until either the message is accepted or the + * specified timeout period elapses. + * + * @param message the {@link Message} to send + * @param timeout the timeout in milliseconds + * + * @return true if the message is sent successfully, + * false if the specified timeout period elapses or + * the send is interrupted + */ + boolean send(Message message, long timeout); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannel.java index cb51a3210c..04fff2374b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -16,7 +16,7 @@ package org.springframework.integration.channel; -import org.springframework.integration.message.BlockingTarget; +import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageSource; /** @@ -24,11 +24,21 @@ import org.springframework.integration.message.MessageSource; * * @author Mark Fisher */ -public interface MessageChannel extends MessageSource, BlockingTarget { +public interface MessageChannel extends MessageSource { /** * Return the name of this channel. */ String getName(); + /** + * Send a {@link Message} to this channel. May throw a RuntimeException for non-recoverable + * errors. Otherwise, if the Message cannot be sent for a non-fatal reason this method will + * return 'false', and if the Message is sent successfully, it will return 'true'. + * + * @param message the Message to send + * @return whether the Message has been sent successfully + */ + boolean send(Message message); + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java index 5dd091822f..5676b13947 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java @@ -62,7 +62,12 @@ public class MessagePublishingErrorHandler implements ErrorHandler { } if (this.errorChannel != null) { try { - this.errorChannel.send(new ErrorMessage(t), this.sendTimeout); + if (this.errorChannel instanceof BlockingChannel && this.sendTimeout >= 0) { + ((BlockingChannel) this.errorChannel).send(new ErrorMessage(t), this.sendTimeout); + } + else { + this.errorChannel.send(new ErrorMessage(t)); + } } catch (Throwable ignore) { // message will be logged only } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java index 182ea83388..f022674ca1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java @@ -21,6 +21,7 @@ import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.integration.channel.BlockingChannel; import org.springframework.integration.channel.ChannelInterceptor; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; @@ -70,7 +71,9 @@ public class MessageStoringInterceptor extends ChannelInterceptorAdapter { } List> storedMessages = this.messageStore.list(); for (Message message : storedMessages) { - if (!channel.send(message, 0)) { + boolean sent = (channel instanceof BlockingChannel) + ? ((BlockingChannel) channel).send(message, 0) : channel.send(message); + if (!sent) { throw new MessagingException("failed to initialize channel from MessageStore"); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java index 30317172c9..34b47fe2a7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java @@ -20,11 +20,11 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; +import org.springframework.integration.channel.BlockingChannel; import org.springframework.integration.channel.ChannelInterceptor; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.BlockingTarget; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.selector.MessageSelector; import org.springframework.util.Assert; @@ -38,7 +38,7 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle { private static final Log logger = LogFactory.getLog(WireTap.class); - private final MessageTarget target; + private final MessageChannel channel; private volatile long timeout = 0; @@ -50,22 +50,22 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle { /** * Create a new wire tap with no {@link MessageSelector}. * - * @param target the MessageTarget to which intercepted messages will be sent + * @param channel the MessageChannel to which intercepted messages will be sent */ - public WireTap(MessageTarget target) { - this(target, null); + public WireTap(MessageChannel channel) { + this(channel, null); } /** * Create a new wire tap with the provided {@link MessageSelector}. * - * @param target the target to which intercepted messages will be sent - * @param selector the selector that must accept a message for it to - * be sent to the intercepting target + * @param channel the channel to which intercepted messages will be sent + * @param selector the selector that must accept a message for it to be + * sent to the intercepting channel */ - public WireTap(MessageTarget target, MessageSelector selector) { - Assert.notNull(target, "target must not be null"); - this.target = target; + public WireTap(MessageChannel channel, MessageSelector selector) { + Assert.notNull(channel, "channel must not be null"); + this.channel = channel; this.selector = selector; } @@ -110,11 +110,11 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle { @Override public Message preSend(Message message, MessageChannel channel) { if (this.running && (this.selector == null || this.selector.accept(message))) { - boolean sent = (this.target instanceof BlockingTarget) - ? ((BlockingTarget) this.target).send(message, this.timeout) - : this.target.send(message); + boolean sent = (this.channel instanceof BlockingChannel && this.timeout >= 0) + ? ((BlockingChannel) this.channel).send(message, this.timeout) + : this.channel.send(message); if (!sent && logger.isWarnEnabled()) { - logger.warn("failed to send message to WireTap target '" + this.target + "'"); + logger.warn("failed to send message to WireTap channel '" + this.channel + "'"); } } return message; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java index 9541e4eb49..34709c10bd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java @@ -95,8 +95,7 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio else { builder.addPropertyReference("source", inputChannel); } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined( - builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, SELECTOR_ATTRIBUTE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, ERROR_HANDLER_ATTRIBUTE); Element interceptorsElement = DomUtils.getChildElementByTagName(element, INTERCEPTORS_ELEMENT); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index 39a165c716..75bf1492f6 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java @@ -81,11 +81,10 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser { adapterBuilder.addPropertyReference("source", source); } if (StringUtils.hasText(channelName)) { - adapterBuilder.addPropertyReference("target", channelName); + adapterBuilder.addPropertyReference("outputChannel", channelName); } else { - adapterBuilder.addPropertyReference("target", - this.createDirectChannel(element, parserContext)); + adapterBuilder.addPropertyReference("outputChannel", this.createDirectChannel(element, parserContext)); } } else if (StringUtils.hasText(target)) { @@ -96,7 +95,7 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser { target = BeanDefinitionReaderUtils.registerWithGeneratedName(invokerBuilder.getBeanDefinition(), parserContext.getRegistry()); } adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(OutboundChannelAdapter.class); - adapterBuilder.addPropertyReference("target", target); + adapterBuilder.addConstructorArgReference(target); if (pollerElement != null) { if (!StringUtils.hasText(channelName)) { throw new ConfigurationException("outbound channel-adapter with a 'poller' requires a 'channel' to poll"); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ResequencerParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ResequencerParser.java index 3d473b7022..c10bf279e9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ResequencerParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ResequencerParser.java @@ -52,7 +52,7 @@ public class ResequencerParser extends AbstractSimpleBeanDefinitionParser { @Override protected void postProcess(BeanDefinitionBuilder builder, Element element) { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, INPUT_CHANNEL_ATTRIBUTE, "source"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index 5c8c52a53e..86e5cab52d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -108,7 +108,7 @@ public abstract class AbstractMethodAnnotationPostProcessor message, MessageTarget target) { - return this.messageExchangeTemplate.send(message, target); + return target.send(message); } public String toString() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index afe16b92e3..a860faf309 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java @@ -16,6 +16,7 @@ package org.springframework.integration.dispatcher; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.SubscribableSource; @@ -25,16 +26,10 @@ import org.springframework.integration.message.SubscribableSource; * * @author Mark Fisher */ -public interface MessageDispatcher extends MessageTarget, SubscribableSource { +public interface MessageDispatcher extends MessageChannel, SubscribableSource { boolean send(Message message); - /** - * Specify the timeout for sending to a target (in milliseconds). - * Note that this value will only be applicable for blocking targets. - */ - void setSendTimeout(long sendTimeout); - boolean subscribe(MessageTarget target); boolean unsubscribe(MessageTarget target); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java index b2e0592859..c9f0629fce 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java @@ -20,7 +20,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.message.BlockingSource; -import org.springframework.integration.message.BlockingTarget; import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.PollableSource; @@ -87,15 +86,6 @@ public class PollingDispatcher implements SchedulableTask, SubscribableSource { this.messageExchangeTemplate.setReceiveTimeout(receiveTimeout); } - /** - * Specify the timeout to use when sending to a target (in milliseconds). - * Note that this value will only be applicable if the target is an instance - * of {@link BlockingTarget}. - */ - public void setSendTimeout(long sendTimeout) { - this.dispatcher.setSendTimeout(sendTimeout); - } - /** * Set the maximum number of messages to receive for each poll. * A non-positive value indicates that polling should repeat as long diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index dab1c1acbd..b93c42ff09 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -22,11 +22,11 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.BeanNameAware; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageSource; -import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.MessagingException; import org.springframework.integration.util.ErrorHandler; @@ -43,7 +43,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist private MessageSource source; - private MessageTarget target; + private MessageChannel outputChannel; private volatile ErrorHandler errorHandler; @@ -71,12 +71,12 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist this.source = source; } - public MessageTarget getTarget() { - return this.target; + public MessageChannel getOutputChannel() { + return this.outputChannel; } - public void setTarget(MessageTarget target) { - this.target = target; + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; } protected ChannelRegistry getChannelRegistry() { 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 bf870b9bf7..014b233ded 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 @@ -21,13 +21,13 @@ import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.MessageChannel; 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.MessagingException; import org.springframework.integration.message.selector.MessageSelector; @@ -82,17 +82,17 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint { return true; } Message reply = buildReplyMessage(result, message.getHeaders()); - MessageTarget replyTarget = this.resolveReplyTarget(message); + MessageChannel replyChannel = this.resolveReplyChannel(message); if (reply instanceof CompositeMessage && this.shouldSplitComposite()) { boolean sentAtLeastOne = false; for (Message nextReply : (CompositeMessage) reply) { - boolean sent = this.sendReplyMessage(nextReply, replyTarget); + boolean sent = this.sendReplyMessage(nextReply, replyChannel); sentAtLeastOne = (sentAtLeastOne || sent); } return sentAtLeastOne; } else { - return this.sendReplyMessage(reply, replyTarget); + return this.sendReplyMessage(reply, replyChannel); } } @@ -112,7 +112,7 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint { return false; } - private boolean sendReplyMessage(Message replyMessage, MessageTarget replyTarget) { + private boolean sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { for (int i = this.interceptors.size() - 1; i >= 0; i--) { EndpointInterceptor interceptor = this.interceptors.get(i); if (interceptor != null) { @@ -122,7 +122,7 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint { } } } - return this.getMessageExchangeTemplate().send(replyMessage, replyTarget); + return this.getMessageExchangeTemplate().send(replyMessage, replyChannel); } private Message buildReplyMessage(Object result, MessageHeaders requestHeaders) { @@ -149,26 +149,30 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint { .build(); } - private MessageTarget resolveReplyTarget(Message requestMessage) { - MessageTarget replyTarget = this.getTarget(); - if (replyTarget == null) { + private MessageChannel resolveReplyChannel(Message requestMessage) { + MessageChannel replyChannel = this.getOutputChannel(); + if (replyChannel == null) { Object returnAddress = requestMessage.getHeaders().getReturnAddress(); if (returnAddress != null) { - if (returnAddress instanceof MessageTarget) { - replyTarget = (MessageTarget) returnAddress; + if (returnAddress instanceof MessageChannel) { + replyChannel = (MessageChannel) returnAddress; } else if (returnAddress instanceof String) { ChannelRegistry channelRegistry = this.getChannelRegistry(); if (channelRegistry != null) { - replyTarget = channelRegistry.lookupChannel((String) returnAddress); + replyChannel = channelRegistry.lookupChannel((String) returnAddress); } } + else { + throw new MessagingException("expected a MessageChannel or String for 'returnAddress', but type is [" + + returnAddress.getClass() + "]"); + } } } - if (replyTarget == null) { - throw new MessagingException("unable to resolve reply target"); + if (replyChannel == null) { + throw new MessagingException("unable to resolve reply channel"); } - return replyTarget; + return replyChannel; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java index be861128ac..995d9465cd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java @@ -34,7 +34,7 @@ public class InboundChannelAdapter extends AbstractEndpoint { @Override protected boolean sendInternal(Message message) { try { - boolean sent = this.getMessageExchangeTemplate().send(message, this.getTarget()); + boolean sent = this.getMessageExchangeTemplate().send(message, this.getOutputChannel()); if (sent && this.getSource() instanceof MessageDeliveryAware) { ((MessageDeliveryAware) this.getSource()).onSend(message); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java index 3172b2bab6..a5d813c3b9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java @@ -17,12 +17,23 @@ package org.springframework.integration.endpoint; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; +import org.springframework.util.Assert; /** * @author Mark Fisher */ public class MessagingBridge extends AbstractRequestReplyEndpoint { + private final MessageTarget target; + + + public MessagingBridge(MessageTarget target) { + Assert.notNull(target, "target must not be null"); + this.target = target; + } + + @Override protected Message handleRequestMessage(Message requestMessage) { return requestMessage; @@ -35,7 +46,7 @@ public class MessagingBridge extends AbstractRequestReplyEndpoint { @Override protected void sendReplyMessage(Message replyMessage, Message requestMessage) { - this.getMessageExchangeTemplate().send(replyMessage, this.getTarget()); + this.target.send(replyMessage); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java index d141c8a892..b294c197d6 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java @@ -18,6 +18,8 @@ package org.springframework.integration.endpoint; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; +import org.springframework.util.Assert; /** * A Channel Adapter implementation for connecting a {@link MessageChannel} @@ -27,9 +29,18 @@ import org.springframework.integration.message.Message; */ public class OutboundChannelAdapter extends AbstractEndpoint { + private final MessageTarget target; + + + public OutboundChannelAdapter(MessageTarget target) { + Assert.notNull(target, "target must not be null"); + this.target = target; + } + + @Override protected boolean sendInternal(Message message) { - return this.getMessageExchangeTemplate().send(message, this.getTarget()); + return this.target.send(message); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java index 156a9f5142..b3a84d51c2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java @@ -20,9 +20,7 @@ import java.util.concurrent.Callable; import java.util.concurrent.FutureTask; import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageExchangeTemplate; -import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.channel.MessageChannel; import org.springframework.util.Assert; /** @@ -42,30 +40,30 @@ public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate { /** - * Send the provided message to the given target. Note that the actual + * Send the provided message to the given channel. Note that the actual * sending occurs asynchronously, so this method will always return * true unless an exception is thrown by the executor. */ @Override - public boolean send(final Message message, final MessageTarget target) { + public boolean send(final Message message, final MessageChannel channel) { this.taskExecutor.execute(new Runnable() { public void run() { - AsyncMessageExchangeTemplate.super.send(message, target); + AsyncMessageExchangeTemplate.super.send(message, channel); } }); return true; } /** - * Send the provided message to the given target and receive the + * Send the provided message to the given channel and receive the * result as an {@link AsyncMessage}. */ @Override @SuppressWarnings("unchecked") - public Message sendAndReceive(final Message request, final MessageTarget target) { + public Message sendAndReceive(final Message request, final MessageChannel channel) { FutureTask> task = new FutureTask>(new Callable>() { public Message call() throws Exception { - return AsyncMessageExchangeTemplate.super.sendAndReceive(request, target); + return AsyncMessageExchangeTemplate.super.sendAndReceive(request, channel); } }); this.taskExecutor.execute(task); @@ -89,15 +87,15 @@ public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate { /** * Receive a Message from the provided source and if not null, - * send it to the given target. Note that the receive and send operations + * send it to the given channel. Note that the receive and send operations * occur asynchronously, so this method will always return true * unless an exception is thrown by the executor. */ @Override - public boolean receiveAndForward(final PollableSource source, final MessageTarget target) { + public boolean receiveAndForward(final PollableSource source, final MessageChannel channel) { this.taskExecutor.execute(new Runnable() { public void run() { - AsyncMessageExchangeTemplate.super.receiveAndForward(source, target); + AsyncMessageExchangeTemplate.super.receiveAndForward(source, channel); } }); return true; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/BlockingTarget.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/BlockingTarget.java index 2db492f0d8..d14b0d0c23 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/BlockingTarget.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/BlockingTarget.java @@ -23,27 +23,6 @@ package org.springframework.integration.message; */ public interface BlockingTarget extends MessageTarget { - /** - * Send a message, blocking indefinitely if necessary. - * - * @param message the {@link Message} to send - * - * @return true if the message is sent successfully, - * false if interrupted - */ - boolean send(Message message); - /** - * Send a message, blocking until either the message is accepted or the - * specified timeout period elapses. - * - * @param message the {@link Message} to send - * @param timeout the timeout in milliseconds - * - * @return true if the message is sent successfully, - * false if the specified timeout period elapses or - * the send is interrupted - */ - boolean send(Message message, long timeout); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java index 50cab9aad8..02cdb42e8b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; +import org.springframework.integration.channel.MessageChannel; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -156,7 +157,7 @@ public final class MessageBuilder { return this.setHeader(MessageHeaders.CORRELATION_ID, correlationId); } - public MessageBuilder setReturnAddress(MessageTarget returnAddress) { + public MessageBuilder setReturnAddress(MessageChannel returnAddress) { return this.setHeader(MessageHeaders.RETURN_ADDRESS, returnAddress); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java index fcb28f0536..1ac3b40a4a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java @@ -23,11 +23,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.message.BlockingSource; -import org.springframework.integration.message.BlockingTarget; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.channel.BlockingChannel; +import org.springframework.integration.channel.MessageChannel; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; @@ -36,9 +33,9 @@ import org.springframework.util.Assert; /** * This is the central class for invoking message exchange operations - * across {@link PollableSource}s and {@link MessageTarget}s. It supports + * across {@link PollableSource}s and {@link MessageChannel}s. It supports * one-way send and receive calls as well as request/reply. Additionally, - * the {@link #receiveAndForward(PollableSource, MessageTarget)} method + * the {@link #receiveAndForward(PollableSource, MessageChannel)} method * plays the role of a polling-consumer while actually sending any * received message to an event-driven consumer. * @@ -143,28 +140,28 @@ public class MessageExchangeTemplate implements InitializingBean { } } - public boolean send(final Message message, final MessageTarget target) { + public boolean send(final Message message, final MessageChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { return (Boolean) txTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - return doSend(message, target); + return doSend(message, channel); } }); } - return this.doSend(message, target); + return this.doSend(message, channel); } - public Message sendAndReceive(final Message request, final MessageTarget target) { + public Message sendAndReceive(final Message request, final MessageChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { return (Message) txTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - return doSendAndReceive(request, target); + return doSendAndReceive(request, channel); } }); } - return this.doSendAndReceive(request, target); + return this.doSendAndReceive(request, channel); } public Message receive(final PollableSource source) { @@ -179,27 +176,27 @@ public class MessageExchangeTemplate implements InitializingBean { return this.doReceive(source); } - public boolean receiveAndForward(final PollableSource source, final MessageTarget target) { + public boolean receiveAndForward(final PollableSource source, final MessageChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { return (Boolean) txTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - return doReceiveAndForward(source, target); + return doReceiveAndForward(source, channel); } }); } - return this.doReceiveAndForward(source, target); + return this.doReceiveAndForward(source, channel); } - private boolean doSend(Message message, MessageTarget target) { - Assert.notNull(target, "target must not be null"); + private boolean doSend(Message message, MessageChannel channel) { + Assert.notNull(channel, "channel must not be null"); long timeout = this.sendTimeout; - boolean sent = (timeout >= 0 && target instanceof BlockingTarget) - ? ((BlockingTarget) target).send(message, timeout) - : target.send(message); + boolean sent = (timeout >= 0 && channel instanceof BlockingChannel) + ? ((BlockingChannel) channel).send(message, timeout) + : channel.send(message); if (!sent && this.logger.isTraceEnabled()) { - this.logger.trace("failed to send message to target '" + target + "' within timeout: " + timeout); + this.logger.trace("failed to send message to channel '" + channel + "' within timeout: " + timeout); } return sent; } @@ -216,23 +213,23 @@ public class MessageExchangeTemplate implements InitializingBean { return message; } - private Message doSendAndReceive(Message request, MessageTarget target) { + private Message doSendAndReceive(Message request, MessageChannel channel) { TemporaryReturnAddress returnAddress = new TemporaryReturnAddress(this.receiveTimeout); request = MessageBuilder.fromMessage(request).setReturnAddress(returnAddress).build(); - if (!this.doSend(request, target)) { + if (!this.doSend(request, channel)) { return null; } return this.doReceive(returnAddress); } - private boolean doReceiveAndForward(PollableSource source, MessageTarget target) { + private boolean doReceiveAndForward(PollableSource source, MessageChannel channel) { Message message = null; try { message = this.doReceive(source); if (message == null) { return false; } - boolean sent = this.doSend(message, target); + boolean sent = this.doSend(message, channel); if (source instanceof MessageDeliveryAware) { if (sent) { ((MessageDeliveryAware) source).onSend(message); @@ -260,7 +257,7 @@ public class MessageExchangeTemplate implements InitializingBean { @SuppressWarnings("unchecked") - private static class TemporaryReturnAddress implements BlockingSource, MessageTarget { + private static class TemporaryReturnAddress implements BlockingSource, MessageChannel { private volatile Message message; @@ -274,6 +271,10 @@ public class MessageExchangeTemplate implements InitializingBean { } + public String getName() { + return "temporaryReplyChannel"; + } + public Message receive() { return this.receive(-1); } 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 d91ed9eb0e..075444defc 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 @@ -126,10 +126,10 @@ public class DefaultMessageBusTests { bus.registerChannel(outputChannel2); endpoint1.setBeanName("testEndpoint1"); endpoint1.setSource(inputChannel); - endpoint1.setTarget(outputChannel1); + endpoint1.setOutputChannel(outputChannel1); endpoint2.setBeanName("testEndpoint2"); endpoint2.setSource(inputChannel); - endpoint2.setTarget(outputChannel2); + endpoint2.setOutputChannel(outputChannel2); bus.registerEndpoint(endpoint1); bus.registerEndpoint(endpoint2); bus.start(); @@ -169,10 +169,10 @@ public class DefaultMessageBusTests { bus.registerChannel(outputChannel2); endpoint1.setBeanName("testEndpoint1"); endpoint1.setSource(inputChannel); - endpoint1.setTarget(outputChannel1); + endpoint1.setOutputChannel(outputChannel1); endpoint2.setBeanName("testEndpoint2"); endpoint2.setSource(inputChannel); - endpoint2.setTarget(outputChannel2); + endpoint2.setOutputChannel(outputChannel2); bus.registerEndpoint(endpoint1); bus.registerEndpoint(endpoint2); 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 4f4aca57d9..715f69515e 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 @@ -62,7 +62,7 @@ public class DirectChannelSubscriptionTests { MethodInvoker invoker = new MessageMappingMethodInvoker(new TestBean(), "handle"); ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(invoker); endpoint.setSource(sourceChannel); - endpoint.setTarget(targetChannel); + endpoint.setOutputChannel(targetChannel); endpoint.setBeanName("testEndpoint"); bus.registerEndpoint(endpoint); bus.start(); @@ -96,7 +96,7 @@ public class DirectChannelSubscriptionTests { } }; endpoint.setSource(sourceChannel); - endpoint.setTarget(targetChannel); + endpoint.setOutputChannel(targetChannel); endpoint.setBeanName("testEndpoint"); bus.registerEndpoint(endpoint); bus.start(); 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 5c572c7973..d859342660 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 @@ -13,7 +13,7 @@ - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index 63f681add3..b27cba07ab 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java @@ -26,6 +26,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.FatalBeanException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.BlockingChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; @@ -52,7 +53,7 @@ public class ChannelParserTests { public void testChannelWithCapacity() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "channelParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("capacityChannel"); + BlockingChannel channel = (BlockingChannel) context.getBean("capacityChannel"); for (int i = 0; i < 10; i++) { boolean result = channel.send(new GenericMessage("test"), 10); assertTrue(result); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java index 4d3c54ee46..35d38b3ecd 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java @@ -22,10 +22,10 @@ import static org.junit.Assert.assertNull; import org.junit.Test; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.MessageSelector; @@ -80,13 +80,13 @@ public class WireTapTests { @Test public void simpleTargetWireTap() { QueueChannel mainChannel = new QueueChannel(); - TestTarget secondaryTarget = new TestTarget(); - mainChannel.addInterceptor(new WireTap(secondaryTarget)); - assertNull(secondaryTarget.getLastMessage()); + TestChannel secondaryChannel = new TestChannel(); + mainChannel.addInterceptor(new WireTap(secondaryChannel)); + assertNull(secondaryChannel.getLastMessage()); Message message = new StringMessage("testing"); mainChannel.send(message); Message original = mainChannel.receive(0); - Message intercepted = secondaryTarget.getLastMessage(); + Message intercepted = secondaryChannel.getLastMessage(); assertNotNull(original); assertNotNull(intercepted); assertEquals(original, intercepted); @@ -124,10 +124,14 @@ public class WireTapTests { } } - private static class TestTarget implements MessageTarget { + private static class TestChannel implements MessageChannel { private volatile Message lastMessage; + public String getName() { + return "testChannel"; + } + public boolean send(Message message) { this.lastMessage = message; return true; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 219b515776..0f5e8d69d7 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -85,7 +85,7 @@ public class AggregatorParserTests { "The AggregatorEndpoint is not injected with the appropriate CompletionStrategy instance", completionStrategy, accessor.getPropertyValue("completionStrategy")); Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel", - outputChannel, accessor.getPropertyValue("target")); + outputChannel, accessor.getPropertyValue("outputChannel")); Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate discard channel", discardChannel, accessor.getPropertyValue("discardChannel")); Assert.assertEquals("The AggregatorEndpoint is not set with the appropriate timeout value", diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index 87e6050799..f016bc87d7 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -72,7 +72,7 @@ public class ResequencerParserTests { public void testDefaultResequencerProperties() { ResequencerEndpoint endpoint = (ResequencerEndpoint) context.getBean("defaultResequencer"); DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint); - Assert.assertNull(accessor.getPropertyValue("target")); + Assert.assertNull(accessor.getPropertyValue("outputChannel")); Assert.assertNull(accessor.getPropertyValue("discardChannel")); Assert.assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", 1000l, accessor.getPropertyValue("sendTimeout")); @@ -97,7 +97,7 @@ public class ResequencerParserTests { MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint); Assert.assertEquals("The ResequencerEndpoint is not injected with the appropriate output channel", - outputChannel, accessor.getPropertyValue("target")); + outputChannel, accessor.getPropertyValue("outputChannel")); Assert.assertEquals("The ResequencerEndpoint is not injected with the appropriate discard channel", discardChannel, accessor.getPropertyValue("discardChannel")); Assert.assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java index 2adc154cd7..4d136a67f2 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java @@ -27,6 +27,7 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.interceptor.WireTap; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -40,12 +41,12 @@ public class WireTapParserTests { public void simpleWireTap() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "wireTapParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("simple"); - WireTapTestTarget target = (WireTapTestTarget) context.getBean("testTarget"); - assertNull(target.getLastMessage()); + MessageChannel mainChannel = (MessageChannel) context.getBean("noSelectors"); + PollableChannel wireTapChannel = (PollableChannel) context.getBean("wireTapChannel"); + assertNull(wireTapChannel.receive(0)); Message original = new StringMessage("test"); - channel.send(original); - Message intercepted = target.getLastMessage(); + mainChannel.send(original); + Message intercepted = wireTapChannel.receive(0); assertNotNull(intercepted); assertEquals(original, intercepted); } @@ -54,12 +55,12 @@ public class WireTapParserTests { public void wireTapWithAcceptingSelector() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "wireTapParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("accepting"); - WireTapTestTarget target = (WireTapTestTarget) context.getBean("testTarget"); - assertNull(target.getLastMessage()); + MessageChannel mainChannel = (MessageChannel) context.getBean("accepting"); + PollableChannel wireTapChannel = (PollableChannel) context.getBean("wireTapChannel"); + assertNull(wireTapChannel.receive(0)); Message original = new StringMessage("test"); - channel.send(original); - Message intercepted = target.getLastMessage(); + mainChannel.send(original); + Message intercepted = wireTapChannel.receive(0); assertNotNull(intercepted); assertEquals(original, intercepted); } @@ -68,12 +69,12 @@ public class WireTapParserTests { public void wireTapWithRejectingSelector() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "wireTapParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("rejecting"); - WireTapTestTarget target = (WireTapTestTarget) context.getBean("testTarget"); - assertNull(target.getLastMessage()); + MessageChannel mainChannel = (MessageChannel) context.getBean("rejecting"); + PollableChannel wireTapChannel = (PollableChannel) context.getBean("wireTapChannel"); + assertNull(wireTapChannel.receive(0)); Message original = new StringMessage("test"); - channel.send(original); - Message intercepted = target.getLastMessage(); + mainChannel.send(original); + Message intercepted = wireTapChannel.receive(0); assertNull(intercepted); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapTestTarget.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapTestTarget.java deleted file mode 100644 index 63e99f4f44..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapTestTarget.java +++ /dev/null @@ -1,39 +0,0 @@ -/* - * Copyright 2002-2008 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.config; - -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; - -/** - * @author Mark Fisher - */ -public class WireTapTestTarget implements MessageTarget { - - private volatile Message lastMessage; - - - public Message getLastMessage() { - return this.lastMessage; - } - - public boolean send(Message message) { - this.lastMessage= message; - return true; - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index 1e0c0cd8b8..8c5715b999 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -41,20 +41,17 @@ public class AggregatorAnnotationTests { ApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithDefaultAnnotation"; - DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, + DirectFieldAccessor accessor = getDirectFieldAccessorForAggregatingHandler(context, endpointName); - Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy); - Assert.assertNull(aggregatingMessageHandlerAccessor.getPropertyValue("target")); - Assert.assertNull(aggregatingMessageHandlerAccessor.getPropertyValue("discardChannel")); - Assert.assertEquals(AggregatorEndpoint.DEFAULT_SEND_TIMEOUT, aggregatingMessageHandlerAccessor - .getPropertyValue("sendTimeout")); - Assert.assertEquals(AggregatorEndpoint.DEFAULT_TIMEOUT, aggregatingMessageHandlerAccessor - .getPropertyValue("timeout")); - Assert.assertEquals(false, aggregatingMessageHandlerAccessor.getPropertyValue("sendPartialResultOnTimeout")); - Assert.assertEquals(AggregatorEndpoint.DEFAULT_REAPER_INTERVAL, aggregatingMessageHandlerAccessor - .getPropertyValue("reaperInterval")); + Assert.assertTrue(accessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy); + Assert.assertNull(accessor.getPropertyValue("outputChannel")); + Assert.assertNull(accessor.getPropertyValue("discardChannel")); + Assert.assertEquals(AggregatorEndpoint.DEFAULT_SEND_TIMEOUT, accessor.getPropertyValue("sendTimeout")); + Assert.assertEquals(AggregatorEndpoint.DEFAULT_TIMEOUT, accessor.getPropertyValue("timeout")); + Assert.assertEquals(false, accessor.getPropertyValue("sendPartialResultOnTimeout")); + Assert.assertEquals(AggregatorEndpoint.DEFAULT_REAPER_INTERVAL, accessor.getPropertyValue("reaperInterval")); Assert.assertEquals(AggregatorEndpoint.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY, - aggregatingMessageHandlerAccessor.getPropertyValue("trackedCorrelationIdCapacity")); + accessor.getPropertyValue("trackedCorrelationIdCapacity")); } @Test @@ -62,22 +59,15 @@ public class AggregatorAnnotationTests { ApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithCustomizedAnnotation"; - DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, - endpointName); - Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy); - Assert.assertEquals(getMessageBus(context).lookupChannel("outputChannel"), aggregatingMessageHandlerAccessor - .getPropertyValue("target")); - Assert.assertEquals(getMessageBus(context).lookupChannel("discardChannel"), aggregatingMessageHandlerAccessor - .getPropertyValue("discardChannel")); - Assert.assertEquals(98765432l, aggregatingMessageHandlerAccessor - .getPropertyValue("sendTimeout")); - Assert.assertEquals(4567890l, aggregatingMessageHandlerAccessor - .getPropertyValue("timeout")); - Assert.assertEquals(true, aggregatingMessageHandlerAccessor.getPropertyValue("sendPartialResultOnTimeout")); - Assert.assertEquals(1234l, aggregatingMessageHandlerAccessor - .getPropertyValue("reaperInterval")); - Assert.assertEquals(42, - aggregatingMessageHandlerAccessor.getPropertyValue("trackedCorrelationIdCapacity")); + DirectFieldAccessor accessor = getDirectFieldAccessorForAggregatingHandler(context, endpointName); + Assert.assertTrue(accessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy); + Assert.assertEquals(getMessageBus(context).lookupChannel("outputChannel"), accessor.getPropertyValue("outputChannel")); + Assert.assertEquals(getMessageBus(context).lookupChannel("discardChannel"), accessor.getPropertyValue("discardChannel")); + Assert.assertEquals(98765432l, accessor.getPropertyValue("sendTimeout")); + Assert.assertEquals(4567890l, accessor.getPropertyValue("timeout")); + Assert.assertEquals(true, accessor.getPropertyValue("sendPartialResultOnTimeout")); + Assert.assertEquals(1234l, accessor.getPropertyValue("reaperInterval")); + Assert.assertEquals(42, accessor.getPropertyValue("trackedCorrelationIdCapacity")); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml index 856de32581..d1a92b74f3 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml @@ -7,36 +7,38 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd"> - + + + + + - + - + - + - + - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java index c0e2d0540c..2320f97001 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java @@ -25,14 +25,10 @@ import static org.easymock.EasyMock.replay; import static org.easymock.EasyMock.reset; import static org.easymock.EasyMock.verify; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.atomic.AtomicBoolean; import org.easymock.IAnswer; import org.junit.Before; @@ -213,57 +209,6 @@ public class BroadcastingDispatcherTests { verify(globalMocks); } - @Test(timeout = 500) - public void multipleTargetsPartialTimeout() throws Exception { - reset(taskExecutorMock); - dispatcher.subscribe(targetMock1); - dispatcher.subscribe(targetMock2); - dispatcher.subscribe(targetMock3); - dispatcher.setSendTimeout(50); - // three threads invoking targets - final CountDownLatch latch = new CountDownLatch(3); - threadedExecutorMock(3); - final AtomicBoolean timingOutStarted = new AtomicBoolean(false); - final AtomicBoolean testNotTimedOut = new AtomicBoolean(false); - expect(targetMock1.send(messageMock)).andAnswer(new IAnswer() { - public Boolean answer() throws Throwable { - latch.countDown(); - return true; - } - }); - expect(targetMock2.send(messageMock)).andAnswer(new IAnswer() { - public Boolean answer() throws Throwable { - latch.countDown(); - return true; - } - }); - /* - * Watch out, this is tricky. The send() method will be invoked but due - * to the faked time out it will never return. Therefore the expectation - * needs to be there, but during the verify it will be called 0 times. - * This is something that EasyMock doesn't support so I've worked around - * it with an AtomicBoolean and a latch. It isn't pretty, but it sort of works - */ - expect(targetMock3.send(messageMock)).andAnswer(new IAnswer() { - public Boolean answer() throws Throwable { - // this should happen - timingOutStarted.compareAndSet(false, true); - latch.countDown(); - // cause timeout here - Thread.sleep(1000); - testNotTimedOut.compareAndSet(false, true); - //in a long running suite this will run until the end, but the test will already be over - return true; - } - }).anyTimes(); - replay(globalMocks); - dispatcher.send(messageMock); - latch.await(); - verify(globalMocks); - assertFalse("Test not timed out properly", testNotTimedOut.get()); - assertTrue("Timing out Runnable not executed", timingOutStarted.get()); - } - @Test public void applySequenceDisabledByDefault() { BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(); @@ -329,21 +274,6 @@ public class BroadcastingDispatcherTests { } } - /* - * expect count calls to the taskExecutorMock.execute and have them run the runnable - * in a new Thread. - */ - private void threadedExecutorMock(int count) { - taskExecutorMock.execute(isA(Runnable.class)); - expectLastCall().andAnswer(new IAnswer() { - public Object answer() throws Throwable { - final Runnable runnable = (Runnable) getCurrentArguments()[0]; - new Thread(runnable).start(); - return null; - } - }).times(count); - } - private static class MessageStoringTestTarget implements MessageTarget { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java index 51d64b1ce5..bf170c5055 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java @@ -40,7 +40,12 @@ public class MessagingBridgeTests { public void simplePassThrough() throws InterruptedException { final CountDownLatch latch = new CountDownLatch(1); DefaultMessageBus bus = new DefaultMessageBus(); - MessagingBridge bridge = new MessagingBridge(); + MessagingBridge bridge = new MessagingBridge(new MessageTarget() { + public boolean send(Message message) { + latch.countDown(); + return true; + } + }); bridge.setBeanName("bridge"); PollableSource source = new PollableSource() { public Message receive() { @@ -50,12 +55,6 @@ public class MessagingBridgeTests { PollingDispatcher poller = new PollingDispatcher(source, new PollingSchedule(1000)); poller.setMaxMessagesPerPoll(1); bridge.setSource(poller); - bridge.setTarget(new MessageTarget() { - public boolean send(Message message) { - latch.countDown(); - return true; - } - }); bus.registerEndpoint(bridge); bus.start(); latch.await(1, TimeUnit.SECONDS); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java index 0acecc9792..c83a47fdc9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java @@ -52,7 +52,7 @@ public class ServiceActivatorEndpointTests { public void outputChannel() { QueueChannel channel = new QueueChannel(1); ServiceActivatorEndpoint endpoint = this.createEndpoint(); - endpoint.setTarget(channel); + endpoint.setOutputChannel(channel); Message message = MessageBuilder.fromPayload("foo").build(); endpoint.send(message); Message reply = channel.receive(0); @@ -65,7 +65,7 @@ public class ServiceActivatorEndpointTests { QueueChannel channel1 = new QueueChannel(1); QueueChannel channel2 = new QueueChannel(1); ServiceActivatorEndpoint endpoint = this.createEndpoint(); - endpoint.setTarget(channel1); + endpoint.setOutputChannel(channel1); Message message = MessageBuilder.fromPayload("foo").setReturnAddress(channel2).build(); endpoint.send(message); Message reply1 = channel1.receive(0); @@ -157,7 +157,7 @@ public class ServiceActivatorEndpointTests { QueueChannel channel = new QueueChannel(1); ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint( new TestNullReplyBean(), "handle"); - endpoint.setTarget(channel); + endpoint.setOutputChannel(channel); Message message = MessageBuilder.fromPayload("foo").build(); endpoint.send(message); assertNull(channel.receive(0)); @@ -169,7 +169,7 @@ public class ServiceActivatorEndpointTests { ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint( new TestNullReplyBean(), "handle"); endpoint.setRequiresReply(true); - endpoint.setTarget(channel); + endpoint.setOutputChannel(channel); Message message = MessageBuilder.fromPayload("foo").build(); endpoint.send(message); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java index 7ae6572c4e..c9ae8b6b8a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java @@ -28,10 +28,10 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.StringMessage; /** @@ -87,7 +87,7 @@ public class GatewayProxyFactoryBeanTests { public void run() { Message input = requestChannel.receive(); StringMessage response = new StringMessage(input.getPayload() + "456"); - ((MessageTarget) input.getHeaders().getReturnAddress()).send(response); + ((MessageChannel) input.getHeaders().getReturnAddress()).send(response); } }).start(); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); @@ -174,7 +174,7 @@ public class GatewayProxyFactoryBeanTests { public void run() { Message input = requestChannel.receive(); StringMessage response = new StringMessage(input.getPayload() + "bar"); - ((MessageTarget) input.getHeaders().getReturnAddress()).send(response); + ((MessageChannel) input.getHeaders().getReturnAddress()).send(response); } }).start(); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); @@ -218,7 +218,7 @@ public class GatewayProxyFactoryBeanTests { public void run() { Message input = requestChannel.receive(); StringMessage response = new StringMessage(input.getPayload() + "bar"); - ((MessageTarget) input.getHeaders().getReturnAddress()).send(response); + ((MessageChannel) input.getHeaders().getReturnAddress()).send(response); } }).start(); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java index 42a8a16ad6..98a88b1a24 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java @@ -102,7 +102,7 @@ public class CorrelationIdTests { MethodInvokingSplitter splitter = new MethodInvokingSplitter( new TestBean(), TestBean.class.getMethod("split", String.class)); SplitterEndpoint endpoint = new SplitterEndpoint(splitter); - endpoint.setTarget(testChannel); + endpoint.setOutputChannel(testChannel); splitter.afterPropertiesSet(); endpoint.send(message); Message reply1 = testChannel.receive(100); 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 596ec48f78..e1d56e34e1 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 @@ -29,6 +29,7 @@ import org.junit.Test; import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.AbstractInOutEndpoint; @@ -69,7 +70,10 @@ public class MessageExchangeTemplateTests { public void testSendWithReturnAddress() throws InterruptedException { final List replies = new ArrayList(3); final CountDownLatch latch = new CountDownLatch(3); - MessageTarget replyTarget = new MessageTarget() { + MessageChannel replyChannel = new MessageChannel() { + public String getName() { + return "testReplyChannel"; + } public boolean send(Message replyMessage) { replies.add((String) replyMessage.getPayload()); latch.countDown(); @@ -77,9 +81,9 @@ public class MessageExchangeTemplateTests { } }; MessageExchangeTemplate template = new MessageExchangeTemplate(); - Message message1 = MessageBuilder.fromPayload("test1").setReturnAddress(replyTarget).build(); - Message message2 = MessageBuilder.fromPayload("test2").setReturnAddress(replyTarget).build(); - Message message3 = MessageBuilder.fromPayload("test3").setReturnAddress(replyTarget).build(); + Message message1 = MessageBuilder.fromPayload("test1").setReturnAddress(replyChannel).build(); + Message message2 = MessageBuilder.fromPayload("test2").setReturnAddress(replyChannel).build(); + Message message3 = MessageBuilder.fromPayload("test3").setReturnAddress(replyChannel).build(); template.send(message1, this.requestChannel); template.send(message2, this.requestChannel); template.send(message3, this.requestChannel);