Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1303 AbstractReplyProducingMessageHandler no longer tries to resolve the replyChannel header value by calling ChannelResolver directly. Instead it simply delegates to MessagingTemplate's new send() method that accepts a "channelName" as a String.

Browse Source
This commit is contained in:
Mark Fisher
2010-07-30 16:14:14 +00:00
parent 7cb3de94d6
commit a3b03bbda7
13 changed files with 114 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -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;
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -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.
*
* <p>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;

9
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java
View File

@@ -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);

8
spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

@@ -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());
}
}

28
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -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;
}
}

62
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

@@ -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 <code>null</code>. In that case, the header value must not also be
* <code>null</code>, 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 <code>null</code>.
*/
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");
}
}
/**

1
spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java
View File

@@ -54,6 +54,7 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl
@Override
public final void onInit() {
super.onInit();
this.processor.setConversionService(this.getConversionService());
}

8
spring-integration-core/src/main/java/org/springframework/integration/router/AbstractChannelNameResolvingMessageRouter.java
View File

@@ -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 + "'");
}

7
spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
View File

@@ -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;

1
spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageProcessingSplitter.java
View File

@@ -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);

7
spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
View File

@@ -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);
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java
View File

@@ -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());
}
}

1
spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java
View File

@@ -66,6 +66,7 @@ public class ApplicationContextMessageBusTests {
}
};
handler.setBeanFactory(context);
handler.afterPropertiesSet();
PollingConsumer endpoint = new PollingConsumer(sourceChannel, handler);
context.registerEndpoint("testEndpoint", endpoint);
context.refresh();