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 410db0b60c..356518a279 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,9 +19,11 @@ import java.util.concurrent.ConcurrentMap; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.context.BeanFactoryChannelResolver; import org.springframework.integration.core.ChannelResolutionException; import org.springframework.integration.core.ChannelResolver; import org.springframework.integration.core.MessageChannel; @@ -76,12 +78,15 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + private volatile ChannelResolver channelResolver; + private volatile MessageChannel discardChannel = new NullChannel(); private boolean sendPartialResultOnExpiry = false; private final ConcurrentMap locks = new ConcurrentHashMap(); + public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(processor); @@ -102,6 +107,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements this(processor, new SimpleMessageStore(0), null, null); } + public void setMessageStore(MessageGroupStore store) { this.messageStore = store; store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { @@ -126,8 +132,13 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements this.outputChannel = outputChannel; } - public void setChannelResolver(ChannelResolver channelResolver) { - super.setChannelResolver(channelResolver); + @Override + protected void onInit() throws Exception { + super.onInit(); + BeanFactory beanFactory = this.getBeanFactory(); + if (this.channelResolver == null && beanFactory != null) { + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); + } } public void setDiscardChannel(MessageChannel discardChannel) { @@ -271,7 +282,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } private MessageChannel resolveReplyChannel(Message requestMessage, MessageChannel defaultOutputChannel) { - ChannelResolver channelResolver = this.getChannelResolver(); MessageChannel replyChannel = defaultOutputChannel; if (replyChannel == null) { Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel(); @@ -281,7 +291,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements else if (replyChannelHeader instanceof String) { Assert.state(channelResolver != null, "ChannelResolver is required for resolving a reply channel by name"); - replyChannel = channelResolver.resolveChannelName((String) replyChannelHeader); + replyChannel = this.channelResolver.resolveChannelName((String) replyChannelHeader); } else if (replyChannelHeader != null) { throw new ChannelResolutionException( 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 91fc6413ce..f96c3d9a24 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 @@ -28,7 +28,6 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.core.convert.ConversionService; import org.springframework.integration.Message; -import org.springframework.integration.core.ChannelResolver; import org.springframework.integration.history.MessageHistoryWriter; import org.springframework.integration.history.NamedComponent; import org.springframework.scheduling.TaskScheduler; @@ -64,8 +63,6 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo private volatile ConversionService conversionService; - private volatile ChannelResolver channelResolver; - public final void setBeanName(String beanName) { this.beanName = beanName; @@ -126,18 +123,6 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo return this.beanFactory; } - protected ChannelResolver getChannelResolver() { - if (this.channelResolver == null && this.beanFactory != null) { - this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory); - } - return this.channelResolver; - } - - protected void setChannelResolver(ChannelResolver channelResolver) { - Assert.notNull(channelResolver, "channelResolver must not be null"); - this.channelResolver = channelResolver; - } - protected TaskScheduler getTaskScheduler() { if (this.taskScheduler == null && this.beanFactory != null) { this.taskScheduler = IntegrationContextUtils.getTaskScheduler(this.beanFactory); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index d7fcba586b..0dd122a509 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -34,7 +34,9 @@ import org.springframework.beans.factory.FactoryBean; import org.springframework.core.convert.ConversionService; import org.springframework.integration.Message; import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.context.BeanFactoryChannelResolver; import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.core.ChannelResolver; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.handler.ArgumentArrayMessageMapper; @@ -68,6 +70,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory private volatile long defaultReplyTimeout = -1; + private volatile ChannelResolver channelResolver; + private volatile TypeConverter typeConverter = new SimpleTypeConverter(); private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); @@ -165,6 +169,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory if (this.initialized) { return; } + BeanFactory beanFactory = this.getBeanFactory(); + if (this.channelResolver == null && beanFactory != null) { + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); + } Class proxyInterface = this.determineServiceInterface(); Method[] methods = proxyInterface.getDeclaredMethods(); for (Method method : methods) { @@ -269,24 +277,30 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory String payloadExpression = null; Map staticHeaders = null; if (gatewayAnnotation != null) { - Assert.state(this.getChannelResolver() != null, "ChannelResolver is required"); String requestChannelName = gatewayAnnotation.requestChannel(); - requestChannel = this.resolveChannel(requestChannel, requestChannelName); + if (StringUtils.hasText(requestChannelName)) { + requestChannel = this.resolveChannelName(requestChannelName); + } String replyChannelName = gatewayAnnotation.replyChannel(); - replyChannel = this.resolveChannel(replyChannel, replyChannelName); + if (StringUtils.hasText(replyChannelName)) { + replyChannel = this.resolveChannelName(replyChannelName); + } requestTimeout = gatewayAnnotation.requestTimeout(); replyTimeout = gatewayAnnotation.replyTimeout(); } else if (methodToChannelMap != null && methodToChannelMap.size() > 0) { - Assert.state(this.getChannelResolver() != null, "ChannelResolver is required"); GatewayMethodDefinition gatewayDefinition = methodToChannelMap.get(method.getName()); if (gatewayDefinition != null) { payloadExpression = gatewayDefinition.getPayloadExpression(); staticHeaders = gatewayDefinition.getStaticHeaders(); String requestChannelName = gatewayDefinition.getRequestChannelName(); - requestChannel = this.resolveChannel(requestChannel, requestChannelName); + if (StringUtils.hasText(requestChannelName)) { + requestChannel = this.resolveChannelName(requestChannelName); + } String replyChannelName = gatewayDefinition.getReplyChannelName(); - replyChannel = this.resolveChannel(replyChannel, replyChannelName); + if (StringUtils.hasText(replyChannelName)) { + replyChannel = this.resolveChannelName(replyChannelName); + } String reqTimeout = gatewayDefinition.getRequestTimeout(); if (StringUtils.hasText(reqTimeout)){ requestTimeout = this.convert(reqTimeout, Long.class); @@ -319,11 +333,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory return gateway; } - private MessageChannel resolveChannel(MessageChannel channel, String channelName) { - if (StringUtils.hasText(channelName)) { - channel = this.getChannelResolver().resolveChannelName(channelName); - Assert.notNull(channel, "failed to resolve channel '" + channelName + "'"); - } + private MessageChannel resolveChannelName(String channelName) { + Assert.state(this.channelResolver != null, "ChannelResolver is required"); + MessageChannel channel = this.channelResolver.resolveChannelName(channelName); + Assert.notNull(channel, "failed to resolve channel '" + channelName + "'"); return channel; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index 7f0223108f..567d62235e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java @@ -19,10 +19,13 @@ package org.springframework.integration.handler; import java.util.HashSet; import java.util.List; +import org.springframework.beans.factory.BeanFactory; import org.springframework.core.Ordered; import org.springframework.integration.Message; import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.context.BeanFactoryChannelResolver; import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.integration.core.ChannelResolver; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.MessageProducer; @@ -75,6 +78,8 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes private volatile int order = Ordered.LOWEST_PRECEDENCE; + private volatile ChannelResolver channelResolver; + private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -111,6 +116,10 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes if (!this.initialized) { Assert.notEmpty(this.handlers, "handler list must not be empty"); this.configureChain(); + BeanFactory beanFactory = this.getBeanFactory(); + if (this.channelResolver == null && beanFactory != null) { + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); + } this.initialized = true; } } @@ -177,8 +186,8 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes replyChannel = (MessageChannel) replyChannelHeader; } else if (replyChannelHeader instanceof String) { - Assert.notNull(getChannelResolver(), "ChannelResolver is required"); - replyChannel = getChannelResolver().resolveChannelName((String) replyChannelHeader); + Assert.notNull(channelResolver, "ChannelResolver is required"); + replyChannel = channelResolver.resolveChannelName((String) replyChannelHeader); } else { throw new MessageHandlingException(message, 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 05237207e1..d5daed9af7 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 @@ -21,10 +21,12 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; +import org.springframework.beans.factory.BeanFactory; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.ConversionServiceFactory; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; +import org.springframework.integration.context.BeanFactoryChannelResolver; import org.springframework.integration.core.ChannelResolutionException; import org.springframework.integration.core.ChannelResolver; import org.springframework.integration.core.MessageChannel; @@ -44,6 +46,8 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract private volatile String suffix; + private volatile ChannelResolver channelResolver; + private volatile boolean ignoreChannelNameResolutionFailures; @@ -53,8 +57,7 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract */ public void setChannelResolver(ChannelResolver channelResolver) { Assert.notNull(channelResolver, "'channelResolver' must not be null"); - super.setChannelResolver(channelResolver); - this.getMessagingTemplate().setChannelResolver(channelResolver); + this.channelResolver = channelResolver; } /** @@ -81,16 +84,18 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract @Override public void onInit() { - Assert.notNull(this.getChannelResolver(), - "either a ChannelResolver or BeanFactory is required"); + BeanFactory beanFactory = this.getBeanFactory(); + if (this.channelResolver == null && beanFactory != null) { + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); + } } private MessageChannel resolveChannelForName(String channelName, Message message) { - Assert.state(this.getChannelResolver() != null, + Assert.state(this.channelResolver != null, "unable to resolve channel names, no ChannelResolver available"); MessageChannel channel = null; try { - channel = this.getChannelResolver().resolveChannelName(channelName); + channel = this.channelResolver.resolveChannelName(channelName); } catch (ChannelResolutionException e) { if (!this.ignoreChannelNameResolutionFailures) {