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

Removed ChannelRegistry. It has been replaced by the ChannelResolver strategy.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-13 02:40:15 +00:00
parent 8419b48dcd
commit f99911cf89
13 changed files with 63 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.bus;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.RendezvousChannel;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
@@ -37,7 +36,7 @@ public class DefaultErrorChannel extends RendezvousChannel {
public DefaultErrorChannel() {
this.addInterceptor(new ErrorLoggingInterceptor());
this.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME);
this.setBeanName(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME);
}

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

@@ -35,7 +35,6 @@ import org.springframework.context.ApplicationEvent;
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.MessageChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.endpoint.MessageEndpoint;
@@ -54,6 +53,9 @@ import org.springframework.util.Assert;
*/
public class DefaultMessageBus implements MessageBus, ApplicationContextAware, ApplicationListener, DisposableBean {
public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel";
private final Log logger = LogFactory.getLog(this.getClass());
private final Set<MessageEndpoint> endpoints = new CopyOnWriteArraySet<MessageEndpoint>();
@@ -233,7 +235,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
}
if (this.taskScheduler instanceof SimpleTaskScheduler) {
((SimpleTaskScheduler) this.taskScheduler).setErrorHandler(
new MessagePublishingErrorHandler(this.lookupChannel(ChannelRegistry.ERROR_CHANNEL_NAME)));
new MessagePublishingErrorHandler(this.lookupChannel(ERROR_CHANNEL_BEAN_NAME)));
}
this.taskScheduler.start();
}

3
org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.bus;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.endpoint.MessageEndpoint;
/**
@@ -25,7 +24,7 @@ import org.springframework.integration.endpoint.MessageEndpoint;
*
* @author Mark Fisher
*/
public interface MessageBus extends ChannelRegistry, Lifecycle {
public interface MessageBus extends Lifecycle {
void registerEndpoint(MessageEndpoint endpoint);

30
org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java
View File

@@ -1,30 +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;
/**
* A strategy interface for registration and lookup of message channels by name.
*
* @author Mark Fisher
*/
public interface ChannelRegistry {
static final String ERROR_CHANNEL_NAME = "errorChannel";
MessageChannel lookupChannel(String channelName);
}

31
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -22,12 +22,13 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.DirectChannel;
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.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumerEndpoint;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
@@ -48,28 +49,20 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
private final ConfigurableBeanFactory beanFactory;
private final ChannelRegistry channelRegistry;
private final ChannelResolver channelResolver;
public ChannelAdapterAnnotationPostProcessor(ConfigurableBeanFactory beanFactory) {
Assert.notNull(beanFactory, "BeanFactory must not be null");
this.beanFactory = beanFactory;
this.channelRegistry = (ChannelRegistry)
this.beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory);
}
public Object postProcess(Object bean, String beanName, Method method, ChannelAdapter annotation) {
Assert.notNull(this.beanFactory, "BeanFactory must not be null");
MessageEndpoint endpoint = null;
String channelName = annotation.value();
MessageChannel channel = this.channelRegistry.lookupChannel(channelName);
if (channel == null) {
DirectChannel directChannel = new DirectChannel();
directChannel.setBeanName(channelName);
this.beanFactory.registerSingleton(channelName, directChannel);
channel = directChannel;
}
MessageChannel channel = this.resolveOrCreateChannel(annotation.value());
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
if (method.getParameterTypes().length == 0 && hasReturnValue(method)) {
MethodInvokingSource source = new MethodInvokingSource();
@@ -94,6 +87,18 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
return bean;
}
private MessageChannel resolveOrCreateChannel(String channelName) {
try {
return this.channelResolver.resolveChannelName(channelName);
}
catch (ChannelResolutionException e) {
DirectChannel directChannel = new DirectChannel();
directChannel.setBeanName(channelName);
this.beanFactory.registerSingleton(channelName, directChannel);
return directChannel;
}
}
private SourcePollingChannelAdapter createInboundChannelAdapter(MethodInvokingSource source, MessageChannel channel, Poller pollerAnnotation) {
Assert.notNull(pollerAnnotation, "The @Poller annotation is required (at method-level) "
+ "when using the @ChannelAdapter annotation with a no-arg method.");

5
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/MessageBusParser.java
View File

@@ -41,7 +41,6 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.bus.DefaultMessageBus;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.bus.MessageBusAwareBeanPostProcessor;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor;
import org.springframework.integration.config.annotation.PublisherAnnotationPostProcessor;
@@ -101,10 +100,10 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
super.doParse(element, parserContext, builder);
String taskSchedulerRef = element.getAttribute(TASK_SCHEDULER_ATTRIBUTE);
TaskExecutor taskExecutor= null;
if (!parserContext.getRegistry().containsBeanDefinition(ChannelRegistry.ERROR_CHANNEL_NAME)) {
if (!parserContext.getRegistry().containsBeanDefinition(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME)) {
RootBeanDefinition errorChannelDef = new RootBeanDefinition(QueueChannel.class);
BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder(
errorChannelDef, ChannelRegistry.ERROR_CHANNEL_NAME);
errorChannelDef, DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME);
BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, parserContext.getRegistry());
}
if (StringUtils.hasText(taskSchedulerRef)) {

9
org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -34,6 +34,8 @@ import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.bus.MessageBus;
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.config.xml.MessageBusParser;
@@ -72,6 +74,8 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor,
private volatile MessageBus messageBus;
private ChannelResolver channelResolver;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
@@ -137,6 +141,7 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor,
public void setBeanFactory(BeanFactory beanFactory) {
this.messageBus = (MessageBus) beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
public void afterPropertiesSet() throws Exception {
@@ -225,12 +230,12 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor,
Assert.state(this.messageBus != null, "MessageBus is required for channel resolution");
String requestChannelName = gatewayAnnotation.requestChannel();
if (StringUtils.hasText(requestChannelName)) {
requestChannel = this.messageBus.lookupChannel(requestChannelName);
requestChannel = this.channelResolver.resolveChannelName(requestChannelName);
Assert.notNull(requestChannel, "failed to resolve request channel '" + requestChannelName + "'");
}
String replyChannelName = gatewayAnnotation.replyChannel();
if (StringUtils.hasText(replyChannelName)) {
replyChannel = this.messageBus.lookupChannel(replyChannelName);
replyChannel = this.channelResolver.resolveChannelName(replyChannelName);
Assert.notNull(replyChannel, "failed to resolve reply channel '" + replyChannelName + "'");
}
requestTimeout = gatewayAnnotation.requestTimeout();

5
org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java
View File

@@ -30,7 +30,6 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -243,8 +242,8 @@ public class DefaultMessageBusTests {
public void consumerSubscribedToErrorChannel() throws InterruptedException {
GenericApplicationContext context = new GenericApplicationContext();
QueueChannel errorChannel = new QueueChannel();
errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME);
context.getBeanFactory().registerSingleton(ChannelRegistry.ERROR_CHANNEL_NAME, errorChannel);
errorChannel.setBeanName(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME);
context.getBeanFactory().registerSingleton(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME, errorChannel);
final CountDownLatch latch = new CountDownLatch(1);
AbstractReplyProducingMessageConsumer consumer = new AbstractReplyProducingMessageConsumer() {
public Message<?> handle(Message<?> message) {

6
org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -24,7 +24,6 @@ import org.junit.Test;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ThreadLocalChannel;
@@ -110,8 +109,9 @@ public class DirectChannelSubscriptionTests {
@Test(expected = MessagingException.class)
public void exceptionThrownFromAnnotatedEndpoint() {
QueueChannel errorChannel = new QueueChannel();
errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME);
context.getBeanFactory().registerSingleton(ChannelRegistry.ERROR_CHANNEL_NAME, errorChannel);
errorChannel.setBeanName(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME);
context.getBeanFactory().registerSingleton(
DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME, errorChannel);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.setBeanFactory(context.getBeanFactory());
postProcessor.afterPropertiesSet();

15
org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java
View File

@@ -24,6 +24,8 @@ import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
@@ -48,7 +50,8 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests
assertTrue(channel instanceof DirectChannel);
MessageBus bus = (MessageBus) this.applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
bus.start();
assertNotNull(bus.lookupChannel(beanName));
BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
assertNotNull(channelResolver.resolveChannelName(beanName));
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertNotNull(adapter);
assertTrue(adapter instanceof SubscribingConsumerEndpoint);
@@ -68,7 +71,8 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests
assertTrue(channel instanceof DirectChannel);
MessageBus bus = (MessageBus) this.applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
bus.start();
assertNotNull(bus.lookupChannel(beanName));
BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
assertNotNull(channelResolver.resolveChannelName(beanName));
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertNotNull(adapter);
assertTrue(adapter instanceof SubscribingConsumerEndpoint);
@@ -87,7 +91,6 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests
PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel");
MessageBus bus = (MessageBus) this.applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
bus.start();
assertNull(bus.lookupChannel(beanName));
Object adapter = this.applicationContext.getBean(beanName);
assertNotNull(adapter);
assertTrue(adapter instanceof SourcePollingChannelAdapter);
@@ -99,4 +102,10 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests
bus.stop();
}
@Test(expected = ChannelResolutionException.class)
public void methodInvokingSourceAdapterIsNotChannel() {
BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
channelResolver.resolveChannelName("methodInvokingSource");
}
}

5
org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java
View File

@@ -36,7 +36,6 @@ import org.springframework.integration.bus.MessageBusInterceptorTests;
import org.springframework.integration.bus.TestMessageBusAwareImpl;
import org.springframework.integration.bus.TestMessageBusStartInterceptor;
import org.springframework.integration.bus.TestMessageBusStopInterceptor;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.scheduling.TaskScheduler;
@@ -54,7 +53,7 @@ public class MessageBusParserTests {
"messageBusWithErrorChannel.xml", this.getClass());
DefaultMessageBus bus = (DefaultMessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
bus.initialize();
MessageChannel channel = bus.lookupChannel(ChannelRegistry.ERROR_CHANNEL_NAME);
MessageChannel channel = bus.lookupChannel(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME);
assertEquals(context.getBean("errorChannel"), channel);
}
@@ -65,7 +64,7 @@ public class MessageBusParserTests {
DefaultMessageBus bus = (DefaultMessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
bus.initialize();
assertNotNull("parser should have created a default error channel",
bus.lookupChannel(ChannelRegistry.ERROR_CHANNEL_NAME));
bus.lookupChannel(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME));
}
@Test

13
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -34,8 +34,8 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.aggregator.SequenceSizeCompletionStrategy;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
/**
@@ -71,9 +71,10 @@ public class AggregatorAnnotationTests {
AbstractMessageAggregator aggregator = this.getAggregator(context, endpointName);
assertTrue(getPropertyValue(aggregator, "completionStrategy")
instanceof SequenceSizeCompletionStrategy);
assertEquals(getMessageBus(context).lookupChannel("outputChannel"),
ChannelResolver channelResolver = new BeanFactoryChannelResolver(context);
assertEquals(channelResolver.resolveChannelName("outputChannel"),
getPropertyValue(aggregator, "outputChannel"));
assertEquals(getMessageBus(context).lookupChannel("discardChannel"),
assertEquals(channelResolver.resolveChannelName("discardChannel"),
getPropertyValue(aggregator, "discardChannel"));
assertEquals(98765432l, getPropertyValue(aggregator, "channelTemplate.sendTimeout"));
assertEquals(4567890l, getPropertyValue(aggregator, "timeout"));
@@ -107,8 +108,4 @@ public class AggregatorAnnotationTests {
return (AbstractMessageAggregator) new DirectFieldAccessor(endpoint).getPropertyValue("consumer");
}
private MessageBus getMessageBus(ApplicationContext context) {
return (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
}
}

8
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -38,6 +38,8 @@ import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.bus.DefaultMessageBus;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
@@ -152,7 +154,8 @@ public class MessagingAnnotationPostProcessorTests {
TargetAnnotationTestBean testBean = new TargetAnnotationTestBean(latch);
postProcessor.postProcessAfterInitialization(testBean, "testBean");
messageBus.start();
MessageChannel testChannel = messageBus.lookupChannel("testChannel");
ChannelResolver channelResolver = new BeanFactoryChannelResolver(context);
MessageChannel testChannel = channelResolver.resolveChannelName("testChannel");
testChannel.send(new StringMessage("foo"));
latch.await(1000, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -397,7 +400,8 @@ public class MessagingAnnotationPostProcessorTests {
ChannelAdapterAnnotationTestBean testBean = new ChannelAdapterAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
messageBus.start();
DirectChannel testChannel = (DirectChannel) messageBus.lookupChannel("testChannel");
ChannelResolver channelResolver = new BeanFactoryChannelResolver(context);
DirectChannel testChannel = (DirectChannel) channelResolver.resolveChannelName("testChannel");
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<Message<?>> receivedMessage = new AtomicReference<Message<?>>();
testChannel.subscribe(new MessageConsumer() {