diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/ApplicationContextMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/ApplicationContextMessageBus.java index ab5d41d0d0..d826d88531 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/ApplicationContextMessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/ApplicationContextMessageBus.java @@ -34,10 +34,8 @@ import org.springframework.context.Lifecycle; import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.scheduling.SimpleTaskScheduler; import org.springframework.integration.scheduling.TaskScheduler; import org.springframework.integration.scheduling.TaskSchedulerAware; import org.springframework.util.Assert; @@ -52,7 +50,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Marius Bogoevici */ -public class ApplicationContextMessageBus implements MessageBus, ApplicationContextAware, ApplicationListener, DisposableBean { +public class ApplicationContextMessageBus implements MessageBus, ChannelResolver, ApplicationContextAware, ApplicationListener, DisposableBean { public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel"; @@ -65,6 +63,8 @@ public class ApplicationContextMessageBus implements MessageBus, ApplicationCont private volatile ApplicationContext applicationContext; + private volatile ChannelResolver channelResolver; + private volatile boolean autoStartup = true; private volatile boolean running; @@ -77,6 +77,7 @@ public class ApplicationContextMessageBus implements MessageBus, ApplicationCont Assert.state(!(applicationContext.getBeanNamesForType(this.getClass()).length > 1), "Only one instance of '" + this.getClass().getSimpleName() + "' is allowed per ApplicationContext."); this.applicationContext = applicationContext; + this.channelResolver = new BeanFactoryChannelResolver(applicationContext); } /** @@ -95,15 +96,8 @@ public class ApplicationContextMessageBus implements MessageBus, ApplicationCont this.autoStartup = autoStartup; } - public MessageChannel lookupChannel(String 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) { - return (MessageChannel) bean; - } - } - return null; + public MessageChannel resolveChannelName(String channelName) { + return this.channelResolver.resolveChannelName(channelName); } public MessageEndpoint lookupEndpoint(String endpointName) { @@ -190,12 +184,6 @@ public class ApplicationContextMessageBus implements MessageBus, ApplicationCont Assert.notNull(this.taskScheduler, "TaskScheduler must not be null"); synchronized (this.lifecycleMonitor) { this.activateEndpoints(); - if (this.taskScheduler instanceof SimpleTaskScheduler) { - ChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext); - MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(channelResolver); - errorHandler.setDefaultErrorChannel(this.lookupChannel(ERROR_CHANNEL_BEAN_NAME)); - ((SimpleTaskScheduler) this.taskScheduler).setErrorHandler(errorHandler); - } this.taskScheduler.start(); } this.running = true; 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 c91ed9654e..06c8aa9d83 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 @@ -37,6 +37,7 @@ import org.springframework.core.task.TaskExecutor; import org.springframework.integration.bus.ApplicationContextMessageBus; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.bus.MessageBusAwareBeanPostProcessor; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor; import org.springframework.integration.config.annotation.PublisherAnnotationPostProcessor; @@ -109,6 +110,11 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser { taskExecutor = this.createTaskExecutor(100, "message-bus-"); BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(SimpleTaskScheduler.class); schedulerBuilder.addConstructorArgValue(taskExecutor); + BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MessagePublishingErrorHandler.class); + errorHandlerBuilder.addPropertyReference("defaultErrorChannel", ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME); + String errorHandlerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName( + errorHandlerBuilder.getBeanDefinition(), parserContext.getRegistry()); + schedulerBuilder.addPropertyReference("errorHandler", errorHandlerBeanName); // TODO: define bean name as a constant elsewhere String TASK_SCHEDULER_BEAN_NAME = "taskScheduler"; BeanDefinitionHolder schedulerHolder = new BeanDefinitionHolder( diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index a226cb025e..9c4133d7e5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java @@ -30,6 +30,10 @@ 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.BeanFactoryChannelResolver; +import org.springframework.integration.channel.ChannelResolutionException; +import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; @@ -47,6 +51,7 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.IntervalTrigger; +import org.springframework.integration.scheduling.SimpleTaskScheduler; import org.springframework.integration.util.TestUtils; /** @@ -222,7 +227,12 @@ public class ApplicationContextMessageBusTests { channelAdapter.setBeanName("testChannel"); context.getBeanFactory().registerSingleton("testChannel", channelAdapter); ApplicationContextMessageBus bus = new ApplicationContextMessageBus(); - bus.setTaskScheduler(TestUtils.createTaskScheduler(10)); + SimpleTaskScheduler taskScheduler = (SimpleTaskScheduler) TestUtils.createTaskScheduler(10); + ChannelResolver channelResolver = new BeanFactoryChannelResolver(context); + MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(channelResolver); + errorHandler.setDefaultErrorChannel(errorChannel); + taskScheduler.setErrorHandler(errorHandler); + bus.setTaskScheduler(taskScheduler); bus.setApplicationContext(context); context.refresh(); bus.start(); @@ -274,18 +284,17 @@ public class ApplicationContextMessageBusTests { context.getBeanFactory().registerSingleton("testChannel", testChannel); ApplicationContextMessageBus messageBus = new ApplicationContextMessageBus(); messageBus.setApplicationContext(context); - MessageChannel lookedUpChannel = messageBus.lookupChannel("testChannel"); + MessageChannel lookedUpChannel = messageBus.resolveChannelName("testChannel"); assertNotNull(testChannel); assertSame(testChannel, lookedUpChannel); } - @Test + @Test(expected = ChannelResolutionException.class) public void lookupNonRegisteredChannel() { GenericApplicationContext context = new GenericApplicationContext(); ApplicationContextMessageBus messageBus = new ApplicationContextMessageBus(); messageBus.setApplicationContext(context); - MessageChannel noSuchChannel = messageBus.lookupChannel("noSuchChannel"); - assertNull(noSuchChannel); + messageBus.resolveChannelName("noSuchChannel"); } @Test 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 ef7e66a7af..d776e20707 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 @@ -51,7 +51,7 @@ public class MessageBusParserTests { ApplicationContext context = new ClassPathXmlApplicationContext( "messageBusWithErrorChannel.xml", this.getClass()); ApplicationContextMessageBus bus = (ApplicationContextMessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); - MessageChannel channel = bus.lookupChannel(ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME); + MessageChannel channel = bus.resolveChannelName(ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME); assertEquals(context.getBean("errorChannel"), channel); } @@ -61,7 +61,7 @@ public class MessageBusParserTests { "messageBusWithDefaults.xml", this.getClass()); ApplicationContextMessageBus bus = (ApplicationContextMessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); assertNotNull("parser should have created a default error channel", - bus.lookupChannel(ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME)); + bus.resolveChannelName(ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME)); } @Test