diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index e3107b344a..4db7f20745 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -20,7 +20,7 @@ import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -41,7 +41,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag private final Log logger = LogFactory.getLog(this.getClass()); @SuppressWarnings("unchecked") - public final void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, + public final void processAndSend(MessageGroup group, MessagingTemplate channelTemplate, MessageChannel outputChannel) { Assert.notNull(group, "MessageGroup must not be null"); Assert.notNull(outputChannel, "'outputChannel' must not be null"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 7c22973f93..f2c524d993 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -19,7 +19,7 @@ import java.util.concurrent.ConcurrentMap; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -51,11 +51,9 @@ import org.springframework.util.Assert; * @author Dave Syer * @since 2.0 */ -public class CorrelatingMessageHandler extends AbstractMessageHandler implements - MessageProducer { +public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer { - private static final Log logger = LogFactory - .getLog(CorrelatingMessageHandler.class); + private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class); public static final long DEFAULT_SEND_TIMEOUT = 1000L; @@ -63,6 +61,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements public static final long DEFAULT_TIMEOUT = 60000L; + private MessageGroupStore messageStore; private final MessageGroupProcessor outputProcessor; @@ -73,7 +72,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private MessageChannel outputChannel; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); private volatile MessageChannel discardChannel = new NullChannel(); @@ -81,23 +80,19 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private final ConcurrentMap locks = new ConcurrentHashMap(); - public CorrelatingMessageHandler(MessageGroupProcessor processor, - MessageGroupStore store, CorrelationStrategy correlationStrategy, - ReleaseStrategy releaseStrategy) { - Assert.notNull(store); + public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, + CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(processor); + Assert.notNull(store); setMessageStore(store); this.outputProcessor = processor; - this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy( - MessageHeaders.CORRELATION_ID) - : correlationStrategy; - this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() - : releaseStrategy; - this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); + this.correlationStrategy = correlationStrategy == null ? + new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID) : correlationStrategy; + this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy; + this.messagingTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); } - public CorrelatingMessageHandler(MessageGroupProcessor processor, - MessageGroupStore store) { + public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) { this(processor, store, null, null); } @@ -138,7 +133,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } public void setSendTimeout(long sendTimeout) { - this.channelTemplate.setSendTimeout(sendTimeout); + this.messagingTemplate.setSendTimeout(sendTimeout); } public void setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) { @@ -162,24 +157,19 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements // TODO: INT-1117 - make the lock global? Object lock = getLock(correlationKey); synchronized (lock) { - MessageGroup group = messageStore.getMessageGroup(correlationKey); - if (group.canAdd(message)) { - group = store(correlationKey, message); - if (releaseStrategy.canRelease(group)) { - if (logger.isDebugEnabled()) { logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } try { - outputProcessor.processAndSend(group, channelTemplate, - this.resolveReplyChannel(message, - this.outputChannel)); - } finally { + outputProcessor.processAndSend(group, messagingTemplate, + this.resolveReplyChannel(message, this.outputChannel)); + } + finally { // Always clean up even if there was an exception // processing messages @@ -187,34 +177,31 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements // The group is complete or else there is no // sequence so there is no more state to track remove(group); - } else { + } + else { // Mark these messages as processed, but do not // remove the group from store mark(group); } - } - - } else if (group.isComplete()) { - + } + else if (group.isComplete()) { try { // If not releasing any messages the group might still // be complete for (Message discard : group.getUnmarked()) { discardChannel.send(discard); } - } finally { + } + finally { remove(group); } - } - - } else { + } + else { discardChannel.send(message); } - } - } private final boolean forceComplete(MessageGroup group) { @@ -227,43 +214,36 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements // last chance for normal completion try { if (releaseStrategy.canRelease(group)) { - outputProcessor.processAndSend(group, channelTemplate, - resolveReplyChannel(group.getOne(), - this.outputChannel)); - } else { + outputProcessor.processAndSend(group, messagingTemplate, + resolveReplyChannel(group.getOne(), this.outputChannel)); + } + else { if (sendPartialResultOnExpiry) { if (logger.isInfoEnabled()) { - logger - .info("Processing partially complete messages for key [" - + correlationKey - + "] to: " - + outputChannel); + logger.info("Processing partially complete messages for key [" + + correlationKey + "] to: " + outputChannel); } - outputProcessor.processAndSend(group, - channelTemplate, resolveReplyChannel(group - .getOne(), this.outputChannel)); - } else { + outputProcessor.processAndSend(group, messagingTemplate, + resolveReplyChannel(group.getOne(), this.outputChannel)); + } + else { if (logger.isInfoEnabled()) { - logger - .info("Discarding partially complete messages for key [" - + correlationKey - + "] to: " - + discardChannel); + logger.info("Discarding partially complete messages for key [" + + correlationKey + "] to: " + discardChannel); } for (Message message : group.getUnmarked()) { discardChannel.send(message); } } } - } finally { + } + finally { remove(group); } return true; } return false; - } - } private Object getLock(Object correlationKey) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index 952eeb26cc..ee5b936613 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -1,6 +1,19 @@ +/* + * Copyright 2002-2010 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.aggregator; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.store.MessageGroup; @@ -13,10 +26,11 @@ import org.springframework.integration.store.MessageGroup; public interface MessageGroupProcessor { /** - * Process the given group and send the resulting message(s) to the output channel using the channel template. + * Process the given group and send the resulting message(s) to the output channel using the messaging template. * Implementations are free to send as little or as many messages based on the invocation as needed. For example an * aggregating processor will send only a single message representing the group, where a resequencing strategy will * send all messages in the group individually. */ - void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel); -} \ No newline at end of file + void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java index fb3b892f85..ba7b98e973 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java @@ -1,6 +1,19 @@ +/* + * Copyright 2002-2010 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.aggregator; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.store.MessageGroup; @@ -15,9 +28,9 @@ import org.springframework.integration.store.MessageGroup; */ public class PassThroughMessageGroupProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { + public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel) { for (Message message : group.getUnmarked()) { - channelTemplate.send(message, outputChannel); + messagingTemplate.send(message, outputChannel); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java index 01be8a2510..4e71a725e7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java @@ -19,7 +19,7 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.store.MessageGroup; @@ -45,13 +45,13 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor this.comparator = comparator; } - public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { + public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel) { Collection> messages = group.getUnmarked(); if (messages.size() > 0) { List> sorted = new ArrayList>(messages); Collections.sort(sorted, comparator); for (Message message : sorted) { - channelTemplate.send(message, outputChannel); + messagingTemplate.send(message, outputChannel); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java index 5c46e8b725..2c8726b626 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java @@ -33,7 +33,7 @@ import org.springframework.expression.spel.SpelParserConfiguration; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; @@ -51,7 +51,7 @@ import org.springframework.util.StringUtils; */ public class MessagePublishingInterceptor implements MethodInterceptor { - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); private volatile ExpressionSource expressionSource; @@ -72,7 +72,7 @@ public class MessagePublishingInterceptor implements MethodInterceptor { } public void setDefaultChannel(MessageChannel defaultChannel) { - this.channelTemplate.setDefaultChannel(defaultChannel); + this.messagingTemplate.setDefaultChannel(defaultChannel); } public void setChannelResolver(ChannelResolver channelResolver) { @@ -136,10 +136,10 @@ public class MessagePublishingInterceptor implements MethodInterceptor { channel = this.channelResolver.resolveChannelName(channelName); } if (channel != null) { - this.channelTemplate.send(message, channel); + this.messagingTemplate.send(message, channel); } else { - this.channelTemplate.send(message); + this.messagingTemplate.send(message); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagingTemplate.java similarity index 98% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/MessagingTemplate.java index 2962890620..97d9a4e01c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagingTemplate.java @@ -48,7 +48,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessageChannelTemplate implements InitializingBean { +public class MessagingTemplate implements InitializingBean { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -79,13 +79,13 @@ public class MessageChannelTemplate implements InitializingBean { * Create a MessageChannelTemplate with no default channel. Note, that one * may be provided by invoking {@link #setDefaultChannel(MessageChannel)}. */ - public MessageChannelTemplate() { + public MessagingTemplate() { } /** * Create a MessageChannelTemplate with the given default channel. */ - public MessageChannelTemplate(MessageChannel defaultChannel) { + public MessagingTemplate(MessageChannel defaultChannel) { this.defaultChannel = defaultChannel; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 0a6709d911..e6b3968bae 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -16,7 +16,7 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageProducer; @@ -32,7 +32,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements private volatile MessageChannel outputChannel; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); public void setOutputChannel(MessageChannel outputChannel) { @@ -40,7 +40,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements } public void setSendTimeout(long sendTimeout) { - this.channelTemplate.setSendTimeout(sendTimeout); + this.messagingTemplate.setSendTimeout(sendTimeout); } @Override @@ -52,7 +52,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements if (message != null) { message.getHeaders().getHistory().addEvent(this); } - return this.channelTemplate.send(message, this.outputChannel); + return this.messagingTemplate.send(message, this.outputChannel); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index a668ed7b7a..9e4893fbb2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -16,7 +16,7 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageSource; @@ -34,7 +34,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint { private volatile MessageChannel outputChannel; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); /** @@ -56,7 +56,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint { * output channel. */ public void setSendTimeout(long sendTimeout) { - this.channelTemplate.setSendTimeout(sendTimeout); + this.messagingTemplate.setSendTimeout(sendTimeout); } @Override @@ -75,7 +75,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint { protected boolean doPoll() { Message message = this.source.receive(); if (message != null) { - return this.channelTemplate.send(message, this.outputChannel); + return this.messagingTemplate.send(message, this.outputChannel); } return false; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java index 689a7eb5df..2c48fb0f25 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -16,7 +16,7 @@ package org.springframework.integration.gateway; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; import org.springframework.integration.core.Message; @@ -52,7 +52,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { private volatile long replyTimeout = 1000; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); private volatile boolean shouldThrowErrors = true; @@ -93,7 +93,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { * @param requestTimeout the timeout value in milliseconds */ public void setRequestTimeout(long requestTimeout) { - this.channelTemplate.setSendTimeout(requestTimeout); + this.messagingTemplate.setSendTimeout(requestTimeout); } /** @@ -104,7 +104,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { */ public void setReplyTimeout(long replyTimeout) { this.replyTimeout = replyTimeout; - this.channelTemplate.setReceiveTimeout(replyTimeout); + this.messagingTemplate.setReceiveTimeout(replyTimeout); } /** @@ -144,7 +144,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { "send is not supported, because no request channel has been configured"); Message message = this.toMessage(object); Assert.notNull(message, "message must not be null"); - if (!this.channelTemplate.send(message, this.requestChannel)) { + if (!this.messagingTemplate.send(message, this.requestChannel)) { throw new MessageDeliveryException(message, "failed to send Message to channel"); } } @@ -153,7 +153,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { this.initializeIfNecessary(); Assert.state(this.replyChannel != null && (this.replyChannel instanceof PollableChannel), "receive is not supported, because no pollable reply channel has been configured"); - Message message = this.channelTemplate.receive((PollableChannel) this.replyChannel); + Message message = this.messagingTemplate.receive((PollableChannel) this.replyChannel); try { return this.fromMessage(message); } @@ -195,7 +195,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { Message reply = null; Throwable error = null; try { - reply = this.channelTemplate.sendAndReceive(message, this.requestChannel); + reply = this.messagingTemplate.sendAndReceive(message, this.requestChannel); if (reply instanceof ErrorMessage) { error = ((ErrorMessage) reply).getPayload(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 098f3935b7..7126d84293 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -17,7 +17,7 @@ package org.springframework.integration.handler; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -34,8 +34,7 @@ import org.springframework.util.Assert; * @author Iwein Fuld * @author Oleg Zhurakousky */ -public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageHandler - implements MessageProducer { +public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageHandler implements MessageProducer { public static final long DEFAULT_SEND_TIMEOUT = 1000; @@ -44,12 +43,12 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa private volatile boolean requiresReply = false; - private final MessageChannelTemplate channelTemplate; + private final MessagingTemplate messagingTemplate; public AbstractReplyProducingMessageHandler() { - this.channelTemplate = new MessageChannelTemplate(); - this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); + this.messagingTemplate = new MessagingTemplate(); + this.messagingTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); } @@ -65,7 +64,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa * Set the timeout for sending reply Messages. */ public void setSendTimeout(long sendTimeout) { - this.channelTemplate.setSendTimeout(sendTimeout); + this.messagingTemplate.setSendTimeout(sendTimeout); } /** @@ -129,7 +128,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa if (logger.isDebugEnabled()) { logger.debug("handler '" + this + "' sending reply Message: " + replyMessage); } - return this.channelTemplate.send(replyMessage, replyChannel); + return this.messagingTemplate.send(replyMessage, replyChannel); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 4c779bcea8..a262fc0d65 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -27,7 +27,7 @@ import org.springframework.core.Ordered; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolutionException; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.Message; @@ -86,7 +86,7 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan private volatile MessageStore messageStore; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); private volatile int order = Ordered.LOWEST_PRECEDENCE; @@ -152,7 +152,7 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan * Set the timeout for sending reply Messages. */ public void setSendTimeout(long sendTimeout) { - this.channelTemplate.setSendTimeout(sendTimeout); + this.messagingTemplate.setSendTimeout(sendTimeout); } /** @@ -241,7 +241,7 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan MessageChannel errorChannel = resolveErrorChannelIfPossible(message); if (errorChannel != null) { ErrorMessage errorMessage = new ErrorMessage(exception); - boolean sent = channelTemplate.send(errorMessage, errorChannel); + boolean sent = messagingTemplate.send(errorMessage, errorChannel); if (!sent && logger.isWarnEnabled()) { logger.warn("Failed to send MessageDeliveryException to error channel.", exception); } @@ -263,7 +263,7 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan private void sendMessageToReplyChannel(Message message) { MessageChannel replyChannel = this.resolveReplyChannel(message); - this.channelTemplate.send(message, replyChannel); + this.messagingTemplate.send(message, replyChannel); } private MessageChannel resolveReplyChannel(Message message) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index 4a16419cb0..178be3da1e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java @@ -19,7 +19,7 @@ package org.springframework.integration.router; import java.util.Collection; import java.util.UUID; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -42,7 +42,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { private volatile boolean applySequence; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); /** @@ -60,7 +60,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { * default, there is no timeout, meaning the send will block indefinitely. */ public void setTimeout(long timeout) { - this.channelTemplate.setSendTimeout(timeout); + this.messagingTemplate.setSendTimeout(timeout); } /** @@ -116,7 +116,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { .setHeader(MessageHeaders.ID, UUID.randomUUID()) .build(); if (channel != null) { - if (this.channelTemplate.send(messageToSend, channel)) { + if (this.messagingTemplate.send(messageToSend, channel)) { sent = true; } else if (!this.ignoreSendFailures) { @@ -128,7 +128,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { } if (!sent) { if (this.defaultOutputChannel != null) { - sent = this.channelTemplate.send(message, this.defaultOutputChannel); + sent = this.messagingTemplate.send(message, this.defaultOutputChannel); } else if (this.resolutionRequired) { throw new MessageDeliveryException(message, diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java index 592cdfb513..8f5de88b50 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java @@ -27,7 +27,7 @@ import java.util.List; import java.util.Map; import org.junit.Test; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; @@ -42,7 +42,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { private final QueueChannel outputChannel = new QueueChannel(1); - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); private final DefaultAggregatingMessageGroupProcessor defaultProcessor = new DefaultAggregatingMessageGroupProcessor(); @@ -107,7 +107,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { Message message = correlatedMessage(1, 1, 1, headers); List> messages = Collections.>singletonList(message); MessageGroup group = new SimpleMessageGroup(messages, 1); - processor.processAndSend(group, channelTemplate, outputChannel); + processor.processAndSend(group, messagingTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); assertEquals("value1", result.getHeaders().get("k1")); @@ -122,7 +122,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { Message message2 = correlatedMessage(1, 2, 2, headers); List> messages = Arrays.>asList(message1, message2); MessageGroup group = new SimpleMessageGroup(messages, 1); - processor.processAndSend(group, channelTemplate, outputChannel); + processor.processAndSend(group, messagingTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); assertEquals("value1", result.getHeaders().get("k1")); @@ -140,7 +140,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { Message message2 = correlatedMessage(1, 2, 2, headers2); List> messages = Arrays.>asList(message1, message2); MessageGroup group = new SimpleMessageGroup(messages, 1); - processor.processAndSend(group, channelTemplate, outputChannel); + processor.processAndSend(group, messagingTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); assertNull(result.getHeaders().get("k1")); @@ -170,7 +170,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { Message message3 = correlatedMessage(1, 3, 3, headers3); List> messages = Arrays.>asList(message1, message2, message3); MessageGroup group = new SimpleMessageGroup(messages, 1); - processor.processAndSend(group, channelTemplate, outputChannel); + processor.processAndSend(group, messagingTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); assertEquals("value1", result.getHeaders().get("only1")); @@ -198,7 +198,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { Message message3 = correlatedMessage(1, 3, 3, headers3); List> messages = Arrays.>asList(message1, message2, message3); MessageGroup group = new SimpleMessageGroup(messages, 1); - processor.processAndSend(group, channelTemplate, outputChannel); + processor.processAndSend(group, messagingTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); assertEquals("valueForAll", result.getHeaders().get("common")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 6c00800815..f89b2478a8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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 @@ -25,7 +25,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -231,18 +231,18 @@ public class AggregatorTests { } private class MultiplyingProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, + public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel) { Integer product = 1; for (Message message : group.getUnmarked()) { product *= (Integer) message.getPayload(); } - channelTemplate.send(MessageBuilder.withPayload(product).build(), outputChannel); + messagingTemplate.send(MessageBuilder.withPayload(product).build(), outputChannel); } } private class NullReturningMessageProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, + public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel) { // noop } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index 5e48743f17..45a3f38381 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -30,7 +30,7 @@ import org.junit.Ignore; import org.junit.Test; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -344,21 +344,19 @@ public class ConcurrentAggregatorTests { private class MultiplyingProcessor implements MessageGroupProcessor { public void processAndSend(MessageGroup group, - MessageChannelTemplate channelTemplate, + MessagingTemplate messagingTemplate, MessageChannel outputChannel) { Integer product = 1; for (Message message : group.getUnmarked()) { product *= (Integer) message.getPayload(); } - channelTemplate.send(MessageBuilder.withPayload(product).build(), - outputChannel); + messagingTemplate.send(MessageBuilder.withPayload(product).build(), outputChannel); } } - private class NullReturningMessageProcessor implements - MessageGroupProcessor { + private class NullReturningMessageProcessor implements MessageGroupProcessor { public void processAndSend(MessageGroup group, - MessageChannelTemplate channelTemplate, + MessagingTemplate messagingTemplate, MessageChannel outputChannel) { // noop } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 7f340cdbcb..aaeaff5236 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -37,7 +37,7 @@ import org.mockito.Mock; import org.mockito.internal.stubbing.answers.DoesNothing; import org.mockito.internal.stubbing.answers.ThrowsException; import org.mockito.runners.MockitoJUnitRunner; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; @@ -75,7 +75,7 @@ public class CorrelatingMessageHandlerTests { ReleaseStrategy); handler.setOutputChannel(outputChannel); doAnswer(new DoesNothing()).when(processor).processAndSend(isA(SimpleMessageGroup.class), - isA(MessageChannelTemplate.class), eq(outputChannel)); + isA(MessagingTemplate.class), eq(outputChannel)); } @Test @@ -94,7 +94,7 @@ public class CorrelatingMessageHandlerTests { verify(correlationStrategy).getCorrelationKey(message1); verify(correlationStrategy).getCorrelationKey(message2); - verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel)); + verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel)); } private void verifyLocks(CorrelatingMessageHandler handler, int lockCount) { @@ -105,7 +105,7 @@ public class CorrelatingMessageHandlerTests { public void bufferCompletesWithException() throws Exception { doAnswer(new ThrowsException(new RuntimeException("Planned test exception"))).when(processor).processAndSend(isA(SimpleMessageGroup.class), - isA(MessageChannelTemplate.class), eq(outputChannel)); + isA(MessagingTemplate.class), eq(outputChannel)); String correlationKey = "key"; Message message1 = testMessage(correlationKey, 1, 2); @@ -124,7 +124,7 @@ public class CorrelatingMessageHandlerTests { verify(correlationStrategy).getCorrelationKey(message1); verify(correlationStrategy).getCorrelationKey(message2); - verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel)); + verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel)); } /* diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index 9e04ae433f..ac691e5237 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -1,3 +1,19 @@ +/* + * Copyright 2002-2010 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.aggregator; import static org.hamcrest.CoreMatchers.is; @@ -19,12 +35,13 @@ import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; + import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.Header; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -44,7 +61,7 @@ public class MethodInvokingMessageGroupProcessorTests { private MessageGroup messageGroupMock; @Mock - private MessageChannelTemplate channelTemplate; + private MessagingTemplate messagingTemplate; @Before public void initializeMessagesUpForProcessing() { @@ -77,9 +94,9 @@ public class MethodInvokingMessageGroupProcessorTests { ArgumentCaptor messageCaptor = ArgumentCaptor.forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, channelTemplate, outputChannel); + processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel); // verify - verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); + verify(messagingTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } @@ -101,9 +118,9 @@ public class MethodInvokingMessageGroupProcessorTests { ArgumentCaptor messageCaptor = ArgumentCaptor.forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, channelTemplate, outputChannel); + processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel); // verify - verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); + verify(messagingTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } @@ -136,9 +153,9 @@ public class MethodInvokingMessageGroupProcessorTests { when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, channelTemplate, outputChannel); + processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel); // verify - verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); + verify(messagingTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } @@ -167,9 +184,9 @@ public class MethodInvokingMessageGroupProcessorTests { when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, channelTemplate, outputChannel); + processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel); // verify - verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); + verify(messagingTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java similarity index 85% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java index 7a18e74673..1106369e2e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessagingTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -43,7 +43,7 @@ import org.springframework.scheduling.support.PeriodicTrigger; /** * @author Mark Fisher */ -public class MessageChannelTemplateTests { +public class MessagingTemplateTests { private TestApplicationContext context = TestUtils.createTestApplicationContext(); @@ -78,7 +78,7 @@ public class MessageChannelTemplateTests { @Test public void send() { - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); QueueChannel channel = new QueueChannel(); template.send(new StringMessage("test"), channel); Message reply = channel.receive(0); @@ -89,7 +89,7 @@ public class MessageChannelTemplateTests { @Test public void sendWithDefaultChannelProvidedBySetter() { QueueChannel channel = new QueueChannel(); - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); template.setDefaultChannel(channel); template.send(new StringMessage("test")); Message reply = channel.receive(0); @@ -100,7 +100,7 @@ public class MessageChannelTemplateTests { @Test public void sendWithDefaultChannelProvidedByConstructor() { QueueChannel channel = new QueueChannel(); - MessageChannelTemplate template = new MessageChannelTemplate(channel); + MessagingTemplate template = new MessagingTemplate(channel); template.send(new StringMessage("test")); Message reply = channel.receive(0); assertNotNull(reply); @@ -111,7 +111,7 @@ public class MessageChannelTemplateTests { public void sendWithExplicitChannelTakesPrecedenceOverDefault() { QueueChannel explicitChannel = new QueueChannel(); QueueChannel defaultChannel = new QueueChannel(); - MessageChannelTemplate template = new MessageChannelTemplate(defaultChannel); + MessagingTemplate template = new MessagingTemplate(defaultChannel); template.send(new StringMessage("test"), explicitChannel); Message reply = explicitChannel.receive(0); assertNotNull(reply); @@ -121,7 +121,7 @@ public class MessageChannelTemplateTests { @Test(expected = IllegalStateException.class) public void sendWithoutChannelArgFailsIfNoDefaultAvailable() { - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); template.send(new StringMessage("test")); } @@ -129,7 +129,7 @@ public class MessageChannelTemplateTests { public void receive() { QueueChannel channel = new QueueChannel(); channel.send(new StringMessage("test")); - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); Message reply = template.receive(channel); assertEquals("test", reply.getPayload()); } @@ -138,7 +138,7 @@ public class MessageChannelTemplateTests { public void receiveWithDefaultChannelProvidedBySetter() { QueueChannel channel = new QueueChannel(); channel.send(new StringMessage("test")); - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); template.setDefaultChannel(channel); Message reply = template.receive(); assertEquals("test", reply.getPayload()); @@ -148,7 +148,7 @@ public class MessageChannelTemplateTests { public void receiveWithDefaultChannelProvidedByConstructor() { QueueChannel channel = new QueueChannel(); channel.send(new StringMessage("test")); - MessageChannelTemplate template = new MessageChannelTemplate(channel); + MessagingTemplate template = new MessagingTemplate(channel); Message reply = template.receive(); assertEquals("test", reply.getPayload()); } @@ -158,7 +158,7 @@ public class MessageChannelTemplateTests { QueueChannel explicitChannel = new QueueChannel(); QueueChannel defaultChannel = new QueueChannel(); explicitChannel.send(new StringMessage("test")); - MessageChannelTemplate template = new MessageChannelTemplate(defaultChannel); + MessagingTemplate template = new MessagingTemplate(defaultChannel); template.setReceiveTimeout(0); Message reply = template.receive(explicitChannel); assertEquals("test", reply.getPayload()); @@ -167,20 +167,20 @@ public class MessageChannelTemplateTests { @Test(expected = IllegalStateException.class) public void receiveWithoutChannelArgFailsIfNoDefaultAvailable() { - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); template.receive(); } @Test(expected = IllegalStateException.class) public void receiveWithNonPollableDefaultFails() { DirectChannel channel = new DirectChannel(); - MessageChannelTemplate template = new MessageChannelTemplate(channel); + MessagingTemplate template = new MessagingTemplate(channel); template.receive(); } @Test public void sendAndReceive() { - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); template.setReceiveTimeout(3000); Message reply = template.sendAndReceive(new StringMessage("test"), this.requestChannel); assertEquals("TEST", reply.getPayload()); @@ -188,7 +188,7 @@ public class MessageChannelTemplateTests { @Test public void sendAndReceiveWithDefaultChannel() { - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); template.setReceiveTimeout(3000); template.setDefaultChannel(this.requestChannel); Message reply = template.sendAndReceive(new StringMessage("test")); @@ -198,7 +198,7 @@ public class MessageChannelTemplateTests { @Test public void sendAndReceiveWithExplicitChannelTakesPrecedenceOverDefault() { QueueChannel defaultChannel = new QueueChannel(); - MessageChannelTemplate template = new MessageChannelTemplate(defaultChannel); + MessagingTemplate template = new MessagingTemplate(defaultChannel); template.setReceiveTimeout(3000); Message message = new StringMessage("test"); Message reply = template.sendAndReceive(message, this.requestChannel); @@ -208,7 +208,7 @@ public class MessageChannelTemplateTests { @Test(expected = IllegalStateException.class) public void sendAndReceiveWithoutChannelArgFailsIfNoDefaultAvailable() { - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); template.sendAndReceive(new StringMessage("test")); } @@ -224,7 +224,7 @@ public class MessageChannelTemplateTests { return true; } }; - MessageChannelTemplate template = new MessageChannelTemplate(); + MessagingTemplate template = new MessagingTemplate(); Message message1 = MessageBuilder.withPayload("test1").setReplyChannel(replyChannel).build(); Message message2 = MessageBuilder.withPayload("test2").setReplyChannel(replyChannel).build(); Message message3 = MessageBuilder.withPayload("test3").setReplyChannel(replyChannel).build(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterParserTests.java index 7876cd4451..5550abeb02 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterParserTests.java @@ -30,7 +30,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.annotation.Router; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -131,7 +131,7 @@ public class RouterParserTests { "routerParserTests.xml", this.getClass()); Object endpoint = context.getBean("routerWithTimeout"); MethodInvokingRouter router = TestUtils.getPropertyValue(endpoint, "handler", MethodInvokingRouter.class); - MessageChannelTemplate template = (MessageChannelTemplate) + MessagingTemplate template = (MessagingTemplate) new DirectFieldAccessor(router).getPropertyValue("channelTemplate"); Long timeout = (Long) new DirectFieldAccessor(template).getPropertyValue("sendTimeout"); assertEquals(new Long(1234), timeout); diff --git a/spring-integration-httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/HttpInvokerInboundGatewayParserTests.java b/spring-integration-httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/HttpInvokerInboundGatewayParserTests.java index b8a75ebe14..c0530311c1 100644 --- a/spring-integration-httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/HttpInvokerInboundGatewayParserTests.java +++ b/spring-integration-httpinvoker/src/test/java/org/springframework/integration/httpinvoker/config/HttpInvokerInboundGatewayParserTests.java @@ -23,7 +23,7 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.httpinvoker.HttpInvokerInboundGateway; @@ -42,7 +42,7 @@ public class HttpInvokerInboundGatewayParserTests { DirectFieldAccessor accessor = new DirectFieldAccessor(gateway); assertEquals(true, accessor.getPropertyValue("expectReply")); assertEquals(channel, accessor.getPropertyValue("requestChannel")); - MessageChannelTemplate template = (MessageChannelTemplate) + MessagingTemplate template = (MessagingTemplate) accessor.getPropertyValue("channelTemplate"); DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); assertEquals(-1L, templateAccessor.getPropertyValue("sendTimeout")); @@ -58,7 +58,7 @@ public class HttpInvokerInboundGatewayParserTests { DirectFieldAccessor accessor = new DirectFieldAccessor(gateway); assertEquals(true, accessor.getPropertyValue("expectReply")); assertEquals(channel, accessor.getPropertyValue("requestChannel")); - MessageChannelTemplate template = (MessageChannelTemplate) + MessagingTemplate template = (MessagingTemplate) accessor.getPropertyValue("channelTemplate"); DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); assertEquals(-1L, templateAccessor.getPropertyValue("sendTimeout")); @@ -74,7 +74,7 @@ public class HttpInvokerInboundGatewayParserTests { DirectFieldAccessor accessor = new DirectFieldAccessor(gateway); assertEquals(false, accessor.getPropertyValue("expectReply")); assertEquals(channel, accessor.getPropertyValue("requestChannel")); - MessageChannelTemplate template = (MessageChannelTemplate) + MessagingTemplate template = (MessagingTemplate) accessor.getPropertyValue("channelTemplate"); DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); assertEquals(123L, templateAccessor.getPropertyValue("sendTimeout")); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java index aa631d8fca..23a19e7c09 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java @@ -1,3 +1,19 @@ +/* + * Copyright 2002-2010 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.jdbc.config; import static org.junit.Assert.assertEquals; @@ -14,7 +30,7 @@ import org.junit.After; import org.junit.Test; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.jdbc.core.namedparam.AbstractSqlParameterSource; @@ -33,7 +49,7 @@ public class JdbcPollingChannelAdapterParserTests { private SimpleJdbcTemplate jdbcTemplate; - private MessageChannelTemplate channelTemplate; + private MessagingTemplate messagingTemplate; private ConfigurableApplicationContext appCtx; @@ -43,7 +59,7 @@ public class JdbcPollingChannelAdapterParserTests { public void testSimpleInboundChannelAdapter(){ setUp("pollingForMapJdbcInboundChannelAdapterTest.xml", getClass()); this.jdbcTemplate.update("insert into item values(1,'',2)"); - Message message = channelTemplate.receive(); + Message message = messagingTemplate.receive(); assertNotNull("No message found ", message); assertTrue("Wrong payload type expected instance of List", message.getPayload() instanceof List); } @@ -53,27 +69,27 @@ public class JdbcPollingChannelAdapterParserTests { public void testSimpleInboundChannelAdapterWithUpdate(){ setUp("pollingForMapJdbcInboundChannelAdapterWithUpdateTest.xml", getClass()); this.jdbcTemplate.update("insert into item values(1,'',2)"); - Message message = channelTemplate.receive(); + Message message = messagingTemplate.receive(); assertNotNull(message); - message = channelTemplate.receive(); - assertNull(channelTemplate.receive()); + message = messagingTemplate.receive(); + assertNull(messagingTemplate.receive()); } @Test public void testSimpleInboundChannelAdapterWithNestedUpdate(){ setUp("pollingForMapJdbcInboundChannelAdapterWithNestedUpdateTest.xml", getClass()); this.jdbcTemplate.update("insert into item values(1,'',2)"); - Message message = channelTemplate.receive(); + Message message = messagingTemplate.receive(); assertNotNull(message); - message = channelTemplate.receive(); - assertNull(channelTemplate.receive()); + message = messagingTemplate.receive(); + assertNull(messagingTemplate.receive()); } @Test public void testExtendedInboundChannelAdapter(){ setUp("pollingWithJdbcOperationsJdbcInboundChannelAdapterTest.xml", getClass()); this.jdbcTemplate.update("insert into item values(1,'',2)"); - Message message = channelTemplate.receive(); + Message message = messagingTemplate.receive(); assertNotNull(message); } @@ -81,7 +97,7 @@ public class JdbcPollingChannelAdapterParserTests { public void testParameterSourceFactoryInboundChannelAdapter(){ setUp("pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml", getClass()); this.jdbcTemplate.update("insert into item values(1,'',2)"); - Message message = channelTemplate.receive(); + Message message = messagingTemplate.receive(); assertNotNull(message); List> list = jdbcTemplate.queryForList("SELECT * FROM item WHERE status=1"); assertEquals(1, list.size()); @@ -92,7 +108,7 @@ public class JdbcPollingChannelAdapterParserTests { public void testParameterSourceInboundChannelAdapter(){ setUp("pollingWithParametersForMapJdbcInboundChannelAdapterTest.xml", getClass()); this.jdbcTemplate.update("insert into item values(1,'',2)"); - Message message = channelTemplate.receive(); + Message message = messagingTemplate.receive(); assertNotNull(message); } @@ -109,14 +125,14 @@ public class JdbcPollingChannelAdapterParserTests { } }); @SuppressWarnings("unchecked") - Message> message = (Message>) channelTemplate.receive(); + Message> message = (Message>) messagingTemplate.receive(); assertNotNull(message); assertEquals(2, message.getPayload().size()); } @After - public void tearDown(){ - if(appCtx != null){ + public void tearDown() { + if(appCtx != null) { appCtx.close(); } } @@ -126,35 +142,34 @@ public class JdbcPollingChannelAdapterParserTests { setupJdbcTemplate(); jdbcTemplate.update("delete from item"); setupTransactionManager(); - setupMessageChannelTemplate(); + setupMessagingTemplate(); } - protected void setupMessageChannelTemplate(){ + protected void setupMessagingTemplate() { PollableChannel pollableChannel = this.appCtx.getBean("target", PollableChannel.class); - this.channelTemplate = new MessageChannelTemplate(pollableChannel); - this.channelTemplate.setReceiveTimeout(500); + this.messagingTemplate = new MessagingTemplate(pollableChannel); + this.messagingTemplate.setReceiveTimeout(500); } - protected void setupJdbcTemplate(){ + protected void setupJdbcTemplate() { this.jdbcTemplate = new SimpleJdbcTemplate(this.appCtx.getBean("dataSource",DataSource.class)); } - protected void setupTransactionManager(){ + protected void setupTransactionManager() { this.transactionManager = this.appCtx.getBean("transactionManager",PlatformTransactionManager.class); } public static class TestSqlParameterSource extends AbstractSqlParameterSource { - public Object getValue(String paramName) - throws IllegalArgumentException { + public Object getValue(String paramName) throws IllegalArgumentException { return 2; } public boolean hasValue(String paramName) { return true; } - + } } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/MailHeaderEnricherTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/MailHeaderEnricherTests.java index 53fc06d39f..e9bd9c885b 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/MailHeaderEnricherTests.java +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/config/MailHeaderEnricherTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -25,7 +25,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.mail.MailHeaders; @@ -48,7 +48,7 @@ public class MailHeaderEnricherTests { @Test public void literalValues() { - MessageChannelTemplate template = new MessageChannelTemplate(literalValuesInput); + MessagingTemplate template = new MessagingTemplate(literalValuesInput); Message result = template.sendAndReceive(new StringMessage("test")); Map headers = result.getHeaders(); assertEquals("test.to", headers.get(MailHeaders.TO)); @@ -61,7 +61,7 @@ public class MailHeaderEnricherTests { @Test public void expressions() { - MessageChannelTemplate template = new MessageChannelTemplate(expressionsInput); + MessagingTemplate template = new MessagingTemplate(expressionsInput); Message result = template.sendAndReceive(new StringMessage("foo")); Map headers = result.getHeaders(); assertEquals("foo.to", headers.get(MailHeaders.TO)); diff --git a/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/config/RmiInboundGatewayParserTests.java b/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/config/RmiInboundGatewayParserTests.java index 6c401c8ae6..2fe6d9536b 100644 --- a/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/config/RmiInboundGatewayParserTests.java +++ b/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/config/RmiInboundGatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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. @@ -24,7 +24,7 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.rmi.RmiInboundGateway; @@ -42,8 +42,8 @@ public class RmiInboundGatewayParserTests { DirectFieldAccessor accessor = new DirectFieldAccessor(gateway); assertEquals(true, accessor.getPropertyValue("expectReply")); assertEquals(channel, accessor.getPropertyValue("requestChannel")); - MessageChannelTemplate template = (MessageChannelTemplate) - accessor.getPropertyValue("channelTemplate"); + MessagingTemplate template = (MessagingTemplate) + accessor.getPropertyValue("messagingTemplate"); DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); assertEquals(-1L, templateAccessor.getPropertyValue("sendTimeout")); assertEquals(-1L, templateAccessor.getPropertyValue("receiveTimeout")); @@ -58,8 +58,8 @@ public class RmiInboundGatewayParserTests { DirectFieldAccessor accessor = new DirectFieldAccessor(gateway); assertEquals(false, accessor.getPropertyValue("expectReply")); assertEquals(channel, accessor.getPropertyValue("requestChannel")); - MessageChannelTemplate template = (MessageChannelTemplate) - accessor.getPropertyValue("channelTemplate"); + MessagingTemplate template = (MessagingTemplate) + accessor.getPropertyValue("messagingTemplate"); DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); assertEquals(123L, templateAccessor.getPropertyValue("sendTimeout")); assertEquals(456L, templateAccessor.getPropertyValue("receiveTimeout")); diff --git a/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceHeaderEnricherTests.java b/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceHeaderEnricherTests.java index 7e15f3fe04..d86cb33f0c 100644 --- a/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceHeaderEnricherTests.java +++ b/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceHeaderEnricherTests.java @@ -25,7 +25,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.StringMessage; @@ -49,7 +49,7 @@ public class WebServiceHeaderEnricherTests { @Test public void literalValue() { - MessageChannelTemplate template = new MessageChannelTemplate(literalValueInput); + MessagingTemplate template = new MessagingTemplate(literalValueInput); Message result = template.sendAndReceive(new StringMessage("foo")); Map headers = result.getHeaders(); assertEquals("http://test", headers.get(WebServiceHeaders.SOAP_ACTION)); @@ -57,7 +57,7 @@ public class WebServiceHeaderEnricherTests { @Test public void expression() { - MessageChannelTemplate template = new MessageChannelTemplate(expressionInput); + MessagingTemplate template = new MessagingTemplate(expressionInput); Message result = template.sendAndReceive(new StringMessage("foo")); Map headers = result.getHeaders(); assertEquals("http://foo", headers.get(WebServiceHeaders.SOAP_ACTION)); diff --git a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java index e8ff347e69..724e01be59 100644 --- a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java +++ b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java @@ -25,7 +25,7 @@ import org.jivesoftware.smack.XMPPConnection; import org.jivesoftware.smack.packet.Message; import org.jivesoftware.smack.packet.Packet; import org.springframework.context.Lifecycle; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.message.MessageBuilder; @@ -63,7 +63,7 @@ import org.springframework.integration.xmpp.XmppHeaders; * @see ChatManager the ChatManager class that * keeps watch over all Chats between the client and any other * participants. - * @see MessageChannelTemplate + * @see MessagingTemplate * handles all interesing operations on any Spring Integration channels. * @see XMPPConnection the XMPPConnection (as * created by {@link XmppConnectionFactory} @@ -72,7 +72,7 @@ public class XmppMessageDrivenEndpoint extends AbstractEndpoint implements Lifec private static final Log logger = LogFactory.getLog(XmppMessageDrivenEndpoint.class); - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); private volatile MessageChannel requestChannel; @@ -80,6 +80,7 @@ public class XmppMessageDrivenEndpoint extends AbstractEndpoint implements Lifec private volatile boolean extractPayload = true; + /** * This will be injected or configured via a xmpp-connection-factory element. * @@ -93,7 +94,7 @@ public class XmppMessageDrivenEndpoint extends AbstractEndpoint implements Lifec * @param requestChannel the channel on which the inbound message should be sent */ public void setRequestChannel(final MessageChannel requestChannel) { - this.channelTemplate.setDefaultChannel(requestChannel); + this.messagingTemplate.setDefaultChannel(requestChannel); this.requestChannel = requestChannel; } @@ -123,7 +124,7 @@ public class XmppMessageDrivenEndpoint extends AbstractEndpoint implements Lifec @Override protected void onInit() throws Exception { - channelTemplate.afterPropertiesSet(); + messagingTemplate.afterPropertiesSet(); xmppConnection.addPacketListener(new PacketListener() { public void processPacket(final Packet packet) { org.jivesoftware.smack.packet.Message message = (org.jivesoftware.smack.packet.Message) packet; @@ -137,7 +138,7 @@ public class XmppMessageDrivenEndpoint extends AbstractEndpoint implements Lifec MessageBuilder messageBuilder = MessageBuilder.withPayload(payload) .setHeader(XmppHeaders.TYPE, xmppMessage.getType()) .setHeader(XmppHeaders.CHAT, chat); - channelTemplate.send(messageBuilder.build(), requestChannel); + messagingTemplate.send(messageBuilder.build(), requestChannel); } } diff --git a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageDrivenEndpoint.java b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageDrivenEndpoint.java index 1e75a1cc79..6047b499aa 100644 --- a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageDrivenEndpoint.java +++ b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageDrivenEndpoint.java @@ -26,7 +26,7 @@ import org.jivesoftware.smack.packet.Presence; import org.springframework.context.Lifecycle; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; @@ -44,11 +44,18 @@ import java.util.Collection; * @since 2.0 */ public class XmppRosterEventMessageDrivenEndpoint extends AbstractEndpoint implements Lifecycle { + private static final Log logger = LogFactory.getLog(XmppRosterEventMessageDrivenEndpoint.class); + + private volatile MessageChannel requestChannel; + private volatile XMPPConnection xmppConnection; + private InboundMessageMapper messageMapper; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + /** * This will be injected or configured via a xmpp-connection-factory element. @@ -63,7 +70,7 @@ public class XmppRosterEventMessageDrivenEndpoint extends AbstractEndpoint imple * @param requestChannel the channel on which the inbound message should be sent */ public void setRequestChannel(final MessageChannel requestChannel) { - this.channelTemplate.setDefaultChannel(requestChannel); + this.messagingTemplate.setDefaultChannel(requestChannel); this.requestChannel = requestChannel; } @@ -88,7 +95,7 @@ public class XmppRosterEventMessageDrivenEndpoint extends AbstractEndpoint imple this.messageMapper = new XmppPresenceMessageMapper(); } - this.channelTemplate.afterPropertiesSet(); + this.messagingTemplate.afterPropertiesSet(); this.xmppConnection.getRoster().addRosterListener(new EventForwardingRosterListener()); } @@ -100,7 +107,7 @@ public class XmppRosterEventMessageDrivenEndpoint extends AbstractEndpoint imple protected void forwardRosterEventMessage(Presence presence) { try { Message msg = this.messageMapper.toMessage(presence); - channelTemplate.send(msg, requestChannel); + messagingTemplate.send(msg, requestChannel); } catch (Exception e) { logger.error("Failed to map packet to message ", e); } @@ -131,4 +138,5 @@ public class XmppRosterEventMessageDrivenEndpoint extends AbstractEndpoint imple forwardRosterEventMessage(presence); } } + } diff --git a/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/config/XmppHeaderEnricherParserTests.java b/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/config/XmppHeaderEnricherParserTests.java index 8b5761543a..5ff0f5d743 100644 --- a/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/config/XmppHeaderEnricherParserTests.java +++ b/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/config/XmppHeaderEnricherParserTests.java @@ -25,16 +25,13 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Value; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.MessagingTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.message.*; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - - - /** * @author Mark Fisher * @author Josh Long @@ -55,8 +52,7 @@ public class XmppHeaderEnricherParserTests { @Test public void to() { - MessageChannelTemplate messageChannelTemplate = new MessageChannelTemplate(); - + MessagingTemplate messagingTemplate = new MessagingTemplate(); output.subscribe(new MessageHandler() { public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, @@ -65,6 +61,7 @@ public class XmppHeaderEnricherParserTests { logger.debug(String.format("%s=%s (class: %s)", h, message.getHeaders().get(h), message.getHeaders().get(h).getClass().toString())); } }); - messageChannelTemplate.send(MessageBuilder.withPayload("foo").build(), input); + messagingTemplate.send(MessageBuilder.withPayload("foo").build(), input); } + }