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

INT-1304 IntegrationObjectSupport no longer controls access to a ChannelResolver

Browse Source
This commit is contained in:
Mark Fisher
2010-07-30 18:18:52 +00:00
parent a3b03bbda7
commit 292c7717f6
5 changed files with 60 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
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(

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

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

35
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -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<String, Object> 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;
}

13
spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
View File

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

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

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