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 f010cacba5..410db0b60c 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 @@ -22,6 +22,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.core.ChannelResolutionException; import org.springframework.integration.core.ChannelResolver; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageProducer; @@ -269,4 +270,29 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements return messageStore.addMessageToGroup(correlationKey, message); } + private MessageChannel resolveReplyChannel(Message requestMessage, MessageChannel defaultOutputChannel) { + ChannelResolver channelResolver = this.getChannelResolver(); + MessageChannel replyChannel = defaultOutputChannel; + if (replyChannel == null) { + Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel(); + if (replyChannelHeader instanceof MessageChannel) { + replyChannel = (MessageChannel) replyChannelHeader; + } + else if (replyChannelHeader instanceof String) { + Assert.state(channelResolver != null, + "ChannelResolver is required for resolving a reply channel by name"); + replyChannel = channelResolver.resolveChannelName((String) replyChannelHeader); + } + else if (replyChannelHeader != null) { + throw new ChannelResolutionException( + "expected a MessageChannel or String for 'replyChannel' header, " + + "but type is [" + replyChannelHeader.getClass() + "]"); + } + } + if (replyChannel == null) { + throw new ChannelResolutionException("unable to resolve reply channel for message: " + requestMessage); + } + return replyChannel; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index cf2a11c66d..91fc6413ce 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -18,6 +18,7 @@ package org.springframework.integration.context; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanFactoryUtils; @@ -36,7 +37,7 @@ import org.springframework.util.StringUtils; /** * A base class that provides convenient access to the bean factory as - * well as {@link ChannelResolver} and {@link TaskScheduler} instances. + * well as {@link TaskScheduler} and {@link ConversionService} instances. * *

This is intended to be used as a base class for internal framework * components whereas code built upon the integration framework should not @@ -59,12 +60,12 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo private volatile BeanFactory beanFactory; - private volatile ChannelResolver channelResolver; - private volatile TaskScheduler taskScheduler; private volatile ConversionService conversionService; + private volatile ChannelResolver channelResolver; + public final void setBeanName(String beanName) { this.beanName = beanName; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index bf7e4d3ae5..2809ad02c9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -72,8 +72,6 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, private volatile boolean readOnly = false; - private volatile BeanFactory beanFactory; - private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -164,7 +162,9 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } public void setBeanFactory(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + if (this.channelResolver == null && beanFactory != null) { + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); + } } public void afterPropertiesSet() { @@ -172,9 +172,6 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, if (this.initialized) { return; } - if (this.channelResolver == null && this.beanFactory != null) { - this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory); - } if (this.transactionManager != null) { TransactionTemplate template = new TransactionTemplate(this.transactionManager); template.setPropagationBehaviorName(this.propagationBehaviorName); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java index 79d43ba0de..27450f1afc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java @@ -90,6 +90,7 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { @Override public final void onInit() { + super.onInit(); if (this.selector instanceof AbstractMessageProcessingSelector) { ((AbstractMessageProcessingSelector) this.selector).setConversionService(this.getConversionService()); } @@ -104,7 +105,7 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { return message; } if (this.discardChannel != null) { - this.sendReplyMessage(message, this.discardChannel); + this.getMessagingTemplate().send(this.discardChannel, message); } if (this.throwExceptionOnRejection) { throw new MessageRejectedException(message); @@ -113,8 +114,9 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { } @Override - protected void handleResult(Object replyMessage, MessageHeaders requestHeaders, MessageChannel replyChannel) { - this.sendReplyMessage((Message) replyMessage, replyChannel); + protected void handleResult(Object replyMessage, MessageHeaders requestHeaders) { + Assert.isInstanceOf(Message.class, replyMessage); + this.sendReplyMessage((Message) replyMessage, requestHeaders.getReplyChannel()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index f6374314a6..674c62b853 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -24,9 +24,6 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingException; import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.core.ChannelResolutionException; -import org.springframework.integration.core.ChannelResolver; -import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.util.Assert; @@ -78,29 +75,4 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im protected abstract void handleMessageInternal(Message message) throws Exception; - protected final MessageChannel resolveReplyChannel(Message requestMessage, MessageChannel defaultOutputChannel) { - ChannelResolver channelResolver = this.getChannelResolver(); - MessageChannel replyChannel = defaultOutputChannel; - if (replyChannel == null) { - Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel(); - if (replyChannelHeader instanceof MessageChannel) { - replyChannel = (MessageChannel) replyChannelHeader; - } - else if (replyChannelHeader instanceof String) { - Assert.state(channelResolver != null, - "ChannelResolver is required for resolving a reply channel by name"); - replyChannel = channelResolver.resolveChannelName((String) replyChannelHeader); - } - else if (replyChannelHeader != null) { - throw new ChannelResolutionException( - "expected a MessageChannel or String for 'replyChannel' header, " + - "but type is [" + replyChannelHeader.getClass() + "]"); - } - } - if (replyChannel == null) { - throw new ChannelResolutionException("unable to resolve reply channel for message: " + requestMessage); - } - return replyChannel; - } - } 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 fc6f147217..4ca6ead28e 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,8 +17,10 @@ package org.springframework.integration.handler; import org.springframework.integration.Message; +import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageHeaders; +import org.springframework.integration.core.ChannelResolutionException; import org.springframework.integration.core.ChannelResolver; import org.springframework.integration.core.MessageBuilder; import org.springframework.integration.core.MessageChannel; @@ -71,7 +73,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa */ public void setChannelResolver(ChannelResolver channelResolver) { Assert.notNull(channelResolver, "'channelResolver' must not be null"); - super.setChannelResolver(channelResolver); + this.messagingTemplate.setChannelResolver(channelResolver); } /** @@ -82,6 +84,20 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa this.requiresReply = requiresReply; } + /** + * Provides access to the {@link MessagingTemplate} for subclasses. + */ + protected MessagingTemplate getMessagingTemplate() { + return this.messagingTemplate; + } + + @Override + protected void onInit() { + if (this.getBeanFactory() != null) { + this.messagingTemplate.setBeanFactory(getBeanFactory()); + } + } + /** * {@inheritDoc} */ @@ -98,14 +114,13 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } return; } - MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel); MessageHeaders requestHeaders = message.getHeaders(); - this.handleResult(result, requestHeaders, replyChannel); + this.handleResult(result, requestHeaders); } - protected void handleResult(Object result, MessageHeaders requestHeaders, MessageChannel replyChannel) { + protected void handleResult(Object result, MessageHeaders requestHeaders) { Message replyMessage = this.createReplyMessage(result, requestHeaders); - this.sendReplyMessage(replyMessage, replyChannel); + this.sendReplyMessage(replyMessage, requestHeaders.getReplyChannel()); } @SuppressWarnings("unchecked") @@ -119,11 +134,44 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa return builder.build(); } - protected void sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { + /** + * Send a reply Message. The 'replyChannelHeaderValue' will be considered only if this handler's + * 'outputChannel' is null. In that case, the header value must not also be + * null, and it must be an instance of either String or {@link MessageChannel}. + * @param replyMessage the reply Message to send + * @param replyChannelHeaderValue the 'replyChannel' header value from the original request + */ + protected final void sendReplyMessage(Message replyMessage, final Object replyChannelHeaderValue) { if (logger.isDebugEnabled()) { logger.debug("handler '" + this + "' sending reply Message: " + replyMessage); } - this.messagingTemplate.send(replyChannel, replyMessage); + MessageChannel outputChannel = this.getOutputChannel(); + if (outputChannel != null) { + this.sendMessage(replyMessage, outputChannel); + } + else if (replyChannelHeaderValue != null) { + this.sendMessage(replyMessage, replyChannelHeaderValue); + } + else { + throw new ChannelResolutionException("no output-channel or replyChannel header available"); + } + } + + /** + * Send the message to the given channel. The channel must be a String or + * {@link MessageChannel} instance, never null. + */ + private void sendMessage(final Message message, final Object channel) { + if (channel instanceof MessageChannel) { + this.messagingTemplate.send((MessageChannel) channel, message); + } + else if (channel instanceof String) { + this.messagingTemplate.send((String) channel, message); + } + else { + throw new MessageDeliveryException(message, + "a non-null reply channel value of type MesssageChannel or String is required"); + } } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java index a579684bde..cf5f1368c2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java @@ -54,6 +54,7 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl @Override public final void onInit() { + super.onInit(); this.processor.setConversionService(this.getConversionService()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractChannelNameResolvingMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractChannelNameResolvingMessageRouter.java index 5632761299..05237207e1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractChannelNameResolvingMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractChannelNameResolvingMessageRouter.java @@ -52,7 +52,9 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract * The default is a BeanFactoryChannelResolver. */ public void setChannelResolver(ChannelResolver channelResolver) { + Assert.notNull(channelResolver, "'channelResolver' must not be null"); super.setChannelResolver(channelResolver); + this.getMessagingTemplate().setChannelResolver(channelResolver); } /** @@ -86,17 +88,17 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract private MessageChannel resolveChannelForName(String channelName, Message message) { Assert.state(this.getChannelResolver() != null, "unable to resolve channel names, no ChannelResolver available"); - MessageChannel channel = null; try { channel = this.getChannelResolver().resolveChannelName(channelName); } catch (ChannelResolutionException e) { - if (!ignoreChannelNameResolutionFailures) + if (!this.ignoreChannelNameResolutionFailures) { throw new MessagingException(message, "failed to resolve channel name '" + channelName + "'", e); + } } - if (channel == null && !ignoreChannelNameResolutionFailures) { + if (channel == null && !this.ignoreChannelNameResolutionFailures) { throw new MessagingException(message, "failed to resolve channel name '" + channelName + "'"); } 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 6dc3aca40d..d7215d9e51 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 @@ -101,6 +101,13 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { return "router"; } + /** + * Provides {@link MessagingTemplate} access for subclasses. + */ + protected MessagingTemplate getMessagingTemplate() { + return this.messagingTemplate; + } + @Override protected void handleMessageInternal(Message message) { boolean sent = false; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageProcessingSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageProcessingSplitter.java index 6547a45845..1094417dd0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageProcessingSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageProcessingSplitter.java @@ -42,6 +42,7 @@ abstract class AbstractMessageProcessingSplitter extends AbstractMessageSplitter @Override public void onInit() { + super.onInit(); ConversionService conversionService = this.getConversionService(); if (conversionService != null && this.messageProcessor instanceof AbstractMessageProcessor) { ((AbstractMessageProcessor) this.messageProcessor).setConversionService(conversionService); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index 291593d160..93ffb7c28b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -24,7 +24,6 @@ import java.util.UUID; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; import org.springframework.integration.core.MessageBuilder; -import org.springframework.integration.core.MessageChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; /** @@ -79,14 +78,14 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess @Override @SuppressWarnings("unchecked") - protected void handleResult(Object result, MessageHeaders requestHeaders, MessageChannel replyChannel) { + protected void handleResult(Object result, MessageHeaders requestHeaders) { if (result instanceof Iterable) { for (Object o : (Iterable) result) { - super.handleResult(o, requestHeaders, replyChannel); + super.handleResult(o, requestHeaders); } } else { - super.handleResult(result, requestHeaders, replyChannel); + super.handleResult(result, requestHeaders); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java index 4e2bab8b2d..28cb8c414d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java @@ -19,7 +19,6 @@ package org.springframework.integration.transformer; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; -import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.util.Assert; @@ -54,6 +53,7 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan @Override protected void onInit() { + super.onInit(); if (this.getBeanFactory() != null && this.transformer instanceof BeanFactoryAware) { ((BeanFactoryAware) this.transformer).setBeanFactory(this.getBeanFactory()); } @@ -73,8 +73,8 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan } @Override - protected void handleResult(Object replyMessage, MessageHeaders requestHeaders, MessageChannel replyChannel) { - this.sendReplyMessage((Message) replyMessage, replyChannel); + protected void handleResult(Object replyMessage, MessageHeaders requestHeaders) { + this.sendReplyMessage((Message) replyMessage, requestHeaders.getReplyChannel()); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index 9d5bbf33ae..4ededf477c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java @@ -66,6 +66,7 @@ public class ApplicationContextMessageBusTests { } }; handler.setBeanFactory(context); + handler.afterPropertiesSet(); PollingConsumer endpoint = new PollingConsumer(sourceChannel, handler); context.registerEndpoint("testEndpoint", endpoint); context.refresh();