From 5a536586e1baf5ba4155f87a78ece7b8b21f4eea Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sat, 11 Oct 2008 17:20:52 +0000 Subject: [PATCH] Removed the 'registerChannel' method from DefaultMessageBus; all channels must be registered within the ApplicationContext. The MessageBusParser now creates the 'errorChannel' if no bean with that name is explicitly defined rather than creating it within the DefaultMessageBus initialization. --- .../integration/bus/DefaultMessageBus.java | 24 +++------------ .../config/xml/MessageBusParser.java | 8 +++++ .../bus/DefaultMessageBusTests.java | 17 ++++------- .../bus/DirectChannelSubscriptionTests.java | 29 ++++++++----------- .../config/MessageBusParserTests.java | 2 +- ...MessagingAnnotationPostProcessorTests.java | 12 ++++---- 6 files changed, 37 insertions(+), 55 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index 81c69521d0..93b16d04ac 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java @@ -20,7 +20,6 @@ import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; @@ -57,8 +56,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A private final Log logger = LogFactory.getLog(this.getClass()); - private final Map channels = new ConcurrentHashMap(); - private final Set endpoints = new CopyOnWriteArraySet(); private final MessageBusInterceptorsList interceptors = new MessageBusInterceptorsList(); @@ -117,9 +114,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A } Assert.notNull(this.applicationContext, "ApplicationContext must not be null"); Assert.notNull(this.taskScheduler, "TaskScheduler must not be null"); - if (this.getErrorChannel() == null) { - this.registerChannel(new DefaultErrorChannel()); - } this.initialized = true; } } @@ -129,24 +123,14 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A } public MessageChannel lookupChannel(String channelName) { - MessageChannel channel = this.channels.get(channelName); - if (channel == null && this.applicationContext != null && this.applicationContext.containsBean(channelName)) { + Assert.notNull(this.applicationContext, "ApplicationContext must not be null"); + if (this.applicationContext.containsBean(channelName)) { Object bean = this.applicationContext.getBean(channelName); if (bean instanceof MessageChannel) { - channel = (MessageChannel) bean; - this.registerChannel(channel); + return (MessageChannel) bean; } } - return channel; - } - - public void registerChannel(MessageChannel channel) { - Assert.notNull(channel, "'channel' must not be null"); - Assert.notNull(channel.getName(), "channel name must not be null"); - this.channels.put(channel.getName(), channel); - if (logger.isInfoEnabled()) { - logger.info("registered channel '" + channel.getName() + "'"); - } + return null; } public void registerEndpoint(MessageEndpoint endpoint) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/MessageBusParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/MessageBusParser.java index e4c3b93964..644e3dc98b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/MessageBusParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/MessageBusParser.java @@ -41,6 +41,8 @@ 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; import org.springframework.integration.scheduling.SimpleTaskScheduler; @@ -99,6 +101,12 @@ 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)) { + RootBeanDefinition errorChannelDef = new RootBeanDefinition(QueueChannel.class); + BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder( + errorChannelDef, ChannelRegistry.ERROR_CHANNEL_NAME); + BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, parserContext.getRegistry()); + } if (StringUtils.hasText(taskSchedulerRef)) { builder.addPropertyReference("taskScheduler", taskSchedulerRef); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java index f16eab267b..8ae8e0eb96 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java @@ -239,15 +239,6 @@ public class DefaultMessageBusTests { new ClassPathXmlApplicationContext("multipleMessageBusBeans.xml", this.getClass()); } - @Test - public void errorChannelRegistration() { - DefaultMessageBus bus = new DefaultMessageBus(); - QueueChannel errorChannel = new QueueChannel(); - errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME); - bus.registerChannel(errorChannel); - assertEquals(errorChannel, bus.getErrorChannel()); - } - @Test public void consumerSubscribedToErrorChannel() throws InterruptedException { GenericApplicationContext context = new GenericApplicationContext(); @@ -275,10 +266,12 @@ public class DefaultMessageBusTests { @Test public void lookupRegisteredChannel() { - DefaultMessageBus messageBus = new DefaultMessageBus(); + GenericApplicationContext context = new GenericApplicationContext(); QueueChannel testChannel = new QueueChannel(); testChannel.setBeanName("testChannel"); - messageBus.registerChannel(testChannel); + context.getBeanFactory().registerSingleton("testChannel", testChannel); + DefaultMessageBus messageBus = new DefaultMessageBus(); + messageBus.setApplicationContext(context); MessageChannel lookedUpChannel = messageBus.lookupChannel("testChannel"); assertNotNull(testChannel); assertSame(testChannel, lookedUpChannel); @@ -286,7 +279,9 @@ public class DefaultMessageBusTests { @Test public void lookupNonRegisteredChannel() { + GenericApplicationContext context = new GenericApplicationContext(); DefaultMessageBus messageBus = new DefaultMessageBus(); + messageBus.setApplicationContext(context); MessageChannel noSuchChannel = messageBus.lookupChannel("noSuchChannel"); assertNull(noSuchChannel); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index e7a45d30a6..293dd0d572 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -43,6 +43,8 @@ import org.springframework.integration.util.TestUtils; */ public class DirectChannelSubscriptionTests { + private GenericApplicationContext context = new GenericApplicationContext(); + private DefaultMessageBus bus = new DefaultMessageBus(); private DirectChannel sourceChannel = new DirectChannel(); @@ -54,14 +56,16 @@ public class DirectChannelSubscriptionTests { public void setupChannels() { sourceChannel.setBeanName("sourceChannel"); targetChannel.setBeanName("targetChannel"); - bus.registerChannel(sourceChannel); - bus.registerChannel(targetChannel); + context.getBeanFactory().registerSingleton("sourceChannel", sourceChannel); + context.getBeanFactory().registerSingleton("targetChannel", targetChannel); + context.getBeanFactory().registerSingleton(MessageBusParser.MESSAGE_BUS_BEAN_NAME, bus); + bus.setApplicationContext(context); bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); } @Test - public void testSendAndReceiveForRegisteredEndpoint() { + public void sendAndReceiveForRegisteredEndpoint() { GenericApplicationContext context = new GenericApplicationContext(); ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(new TestBean(), "handle"); serviceActivator.setOutputChannel(targetChannel); @@ -76,10 +80,7 @@ public class DirectChannelSubscriptionTests { } @Test - public void testSendAndReceiveForAnnotatedEndpoint() { - GenericApplicationContext context = new GenericApplicationContext(); - bus.setApplicationContext(context); - context.getBeanFactory().registerSingleton(MessageBusParser.MESSAGE_BUS_BEAN_NAME, bus); + public void sendAndReceiveForAnnotatedEndpoint() { MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); @@ -92,11 +93,8 @@ public class DirectChannelSubscriptionTests { bus.stop(); } - @Test(expected=RuntimeException.class) - public void testExceptionThrownFromRegisteredEndpoint() { - QueueChannel errorChannel = new QueueChannel(); - errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME); - bus.registerChannel(errorChannel); + @Test(expected = MessagingException.class) + public void exceptionThrownFromRegisteredEndpoint() { AbstractReplyProducingMessageConsumer consumer = new AbstractReplyProducingMessageConsumer() { public Message handle(Message message) { throw new RuntimeException("intentional test failure"); @@ -109,11 +107,8 @@ public class DirectChannelSubscriptionTests { this.sourceChannel.send(new StringMessage("foo")); } - @Test(expected=MessagingException.class) - public void testExceptionThrownFromAnnotatedEndpoint() { - GenericApplicationContext context = new GenericApplicationContext(); - bus.setApplicationContext(context); - context.getBeanFactory().registerSingleton(MessageBusParser.MESSAGE_BUS_BEAN_NAME, bus); + @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); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index 6660f594a7..0211c0b741 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java @@ -61,7 +61,7 @@ public class MessageBusParserTests { "messageBusWithDefaults.xml", this.getClass()); DefaultMessageBus bus = (DefaultMessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); bus.initialize(); - assertNotNull("bus should have created a default error channel", bus.getErrorChannel()); + assertNotNull("parser should have created a default error channel", bus.getErrorChannel()); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index 7d198afd65..1ed91d7b28 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -62,13 +62,13 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testServiceActivatorAnnotation() { GenericApplicationContext context = new GenericApplicationContext(); + QueueChannel inputChannel = new QueueChannel(); + inputChannel.setBeanName("inputChannel"); + context.getBeanFactory().registerSingleton("inputChannel", inputChannel); DefaultMessageBus messageBus = new DefaultMessageBus(); context.getBeanFactory().registerSingleton( MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); - QueueChannel inputChannel = new QueueChannel(); - inputChannel.setBeanName("inputChannel"); - messageBus.registerChannel(inputChannel); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); @@ -181,13 +181,13 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testChannelRegistryAwareBean() { GenericApplicationContext context = new GenericApplicationContext(); + QueueChannel inputChannel = new QueueChannel(); + inputChannel.setBeanName("inputChannel"); + context.getBeanFactory().registerSingleton("inputChannel", inputChannel); DefaultMessageBus messageBus = new DefaultMessageBus(); context.getBeanFactory().registerSingleton( MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); - QueueChannel inputChannel = new QueueChannel(); - inputChannel.setBeanName("inputChannel"); - messageBus.registerChannel(inputChannel); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet();