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