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

Removed ChannelRegistryAware. The AbstractReplyProducingMessageConsumer now uses a BeanFactoryChannelResolver and is therefore now BeanFactoryAware. The annotation post-processors set the BeanFactoryChannelResolver when creating such a consumer (since it is not created within the ApplicationContext in that case).

Browse Source
This commit is contained in:
Mark Fisher
2008-10-13 00:40:19 +00:00
parent d4271811f3
commit a8d25f6171
13 changed files with 62 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
View File

@@ -36,7 +36,6 @@ import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.endpoint.MessageEndpoint;
@@ -177,9 +176,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
private void activateEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
if (endpoint instanceof ChannelRegistryAware) {
((ChannelRegistryAware) endpoint).setChannelRegistry(this);
}
if (endpoint instanceof TaskSchedulerAware) {
((TaskSchedulerAware) endpoint).setTaskScheduler(this.taskScheduler);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.bus;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.util.Assert;
/**
@@ -51,9 +50,6 @@ public class MessageBusAwareBeanPostProcessor implements BeanPostProcessor {
if (bean instanceof MessageBusAware) {
((MessageBusAware) bean).setMessageBus(this.messageBus);
}
if (bean instanceof ChannelRegistryAware) {
((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus);
}
return bean;
}

29
org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistryAware.java
View File

@@ -1,29 +0,0 @@
/*
* Copyright 2002-2008 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.channel;
/**
* Interface to be implemented by components that need access to the
* {@link ChannelRegistry}.
*
* @author Mark Fisher
*/
public interface ChannelRegistryAware {
void setChannelRegistry(ChannelRegistry channelRegistry);
}

22
org.springframework.integration/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -18,12 +18,11 @@ package org.springframework.integration.config;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
@@ -40,10 +39,12 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class ConsumerEndpointFactoryBean implements FactoryBean, ChannelRegistryAware, BeanFactoryAware, InitializingBean {
public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, InitializingBean {
private final MessageConsumer consumer;
private volatile String beanName;
private volatile String inputChannelName;
private volatile Trigger trigger;
@@ -73,14 +74,12 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, ChannelRegistry
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
public void setBeanName(String beanName) {
this.beanName = beanName;
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
if (this.consumer instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.consumer).setChannelRegistry(channelRegistry);
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public void setTrigger(Trigger trigger) {
@@ -141,11 +140,12 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, ChannelRegistry
return;
}
Assert.isTrue(this.beanFactory.containsBean(this.inputChannelName),
"no such input channel '" + this.inputChannelName + "'");
"no such input channel '" + this.inputChannelName + "' for endpoint '" + this.beanName + "'");
MessageChannel channel = (MessageChannel)
this.beanFactory.getBean(this.inputChannelName, MessageChannel.class);
if (channel instanceof SubscribableChannel) {
Assert.isNull(trigger, "A trigger should not be specified when using a SubscribableChannel");
Assert.isNull(trigger, "A trigger should not be specified for endpoint '" + this.beanName
+ "', since '" + this.inputChannelName + "' is a SubscribableChannel (not pollable).");
this.endpoint = new SubscribingConsumerEndpoint(
this.consumer, (SubscribableChannel) channel);
}

20
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -24,12 +24,11 @@ import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.endpoint.AbstractMessageConsumer;
import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.endpoint.MessageEndpoint;
@@ -53,22 +52,18 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
protected final GenericBeanFactoryAccessor beanFactoryAccessor;
protected final ChannelRegistry channelRegistry;
protected final ChannelResolver channelResolver;
public AbstractMethodAnnotationPostProcessor(ListableBeanFactory beanFactory) {
Assert.notNull(beanFactory, "BeanFactory must not be null");
this.beanFactoryAccessor = new GenericBeanFactoryAccessor(beanFactory);
this.channelRegistry = this.beanFactoryAccessor.getBean(
MessageBusParser.MESSAGE_BUS_BEAN_NAME, ChannelRegistry.class);
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
public Object postProcess(Object bean, String beanName, Method method, T annotation) {
MessageConsumer consumer = this.createConsumer(bean, method, annotation);
if (consumer instanceof ChannelRegistryAware) {
((ChannelRegistryAware) consumer).setChannelRegistry(this.channelRegistry);
}
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
MessageEndpoint endpoint = this.createEndpoint(consumer, annotation, pollerAnnotation);
if (endpoint != null) {
@@ -94,8 +89,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
MessageEndpoint endpoint = null;
String inputChannelName = (String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(inputChannelName)) {
MessageChannel inputChannel = this.channelRegistry.lookupChannel(inputChannelName);
Assert.notNull(inputChannel, "unable to resolve inputChannel '" + inputChannelName + "'");
MessageChannel inputChannel = this.channelResolver.resolveChannelName(inputChannelName);
Assert.notNull(inputChannel, "failed to resolve inputChannel '" + inputChannelName + "'");
if (consumer instanceof AbstractMessageConsumer) {
if (inputChannel instanceof PollableChannel) {
PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
@@ -119,10 +114,11 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
if (consumer instanceof AbstractReplyProducingMessageConsumer) {
String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = this.channelRegistry.lookupChannel(outputChannelName);
MessageChannel outputChannel = this.channelResolver.resolveChannelName(outputChannelName);
Assert.notNull(outputChannel, "unable to resolve outputChannel '" + outputChannelName + "'");
((AbstractReplyProducingMessageConsumer) consumer).setOutputChannel(outputChannel);
}
((AbstractReplyProducingMessageConsumer) consumer).setChannelResolver(this.channelResolver);
}
}
return endpoint;

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -50,8 +50,8 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
this.configureCompletionStrategy(bean, aggregator);
String discardChannelName = annotation.discardChannel();
if (StringUtils.hasText(discardChannelName)) {
MessageChannel discardChannel = this.channelRegistry.lookupChannel(discardChannelName);
Assert.notNull(discardChannel, "unable to resolve discardChannel '" + discardChannelName + "'");
MessageChannel discardChannel = this.channelResolver.resolveChannelName(discardChannelName);
Assert.notNull(discardChannel, "failed to resolve discardChannel '" + discardChannelName + "'");
aggregator.setDiscardChannel(discardChannel);
}
aggregator.setSendTimeout(annotation.sendTimeout());

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -40,7 +40,6 @@ import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.stereotype.Component;
@@ -118,9 +117,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
}
}
});
if (bean instanceof ChannelRegistryAware) {
((ChannelRegistryAware) bean).setChannelRegistry(messageBus);
}
return bean;
}

5
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java
View File

@@ -20,7 +20,6 @@ import java.lang.reflect.Method;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.router.MethodInvokingRouter;
@@ -42,10 +41,10 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP
@Override
protected MessageConsumer createConsumer(Object bean, Method method, Router annotation) {
MethodInvokingRouter router = new MethodInvokingRouter(bean, method);
router.setChannelResolver(new BeanFactoryChannelResolver(this.beanFactoryAccessor.getBeanFactory()));
router.setChannelResolver(this.channelResolver);
String defaultOutputChannelName = annotation.defaultOutputChannel();
if (StringUtils.hasText(defaultOutputChannelName)) {
MessageChannel defaultOutputChannel = this.channelRegistry.lookupChannel(defaultOutputChannelName);
MessageChannel defaultOutputChannel = this.channelResolver.resolveChannelName(defaultOutputChannelName);
Assert.notNull(defaultOutputChannel, "unable to resolve defaultOutputChannel '" + defaultOutputChannelName + "'");
router.setDefaultOutputChannel(defaultOutputChannel);
}

26
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java
View File

@@ -19,8 +19,10 @@ package org.springframework.integration.endpoint;
import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.message.CompositeMessage;
@@ -38,14 +40,14 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer implements ChannelRegistryAware {
public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer implements BeanFactoryAware {
public static final long DEFAULT_SEND_TIMEOUT = 1000;
private MessageChannel outputChannel;
private volatile ChannelRegistry channelRegistry;
private volatile ChannelResolver channelResolver;
private volatile MessageSelector selector;
@@ -75,8 +77,9 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
this.channelTemplate.setSendTimeout(sendTimeout);
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
public void setChannelResolver(ChannelResolver channelResolver) {
Assert.notNull(channelResolver, "channelResolver must not be null");
this.channelResolver = channelResolver;
}
public void setSelector(MessageSelector selector) {
@@ -87,6 +90,11 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
this.requiresReply = requiresReply;
}
public void setBeanFactory(BeanFactory beanFactory) {
if (this.channelResolver == null) {
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
}
@Override
protected final void onMessageInternal(Message<?> message) {
@@ -175,9 +183,9 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
replyChannel = (MessageChannel) returnAddress;
}
else if (returnAddress instanceof String) {
Assert.state(this.channelRegistry != null,
"ChannelRegistry is required for resolving a reply channel by name");
replyChannel = this.channelRegistry.lookupChannel((String) returnAddress);
Assert.state(this.channelResolver != null,
"ChannelResolver is required for resolving a reply channel by name");
replyChannel = this.channelResolver.resolveChannelName((String) returnAddress);
}
else {
throw new MessagingException("expected a MessageChannel or String for 'returnAddress', but type is ["