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

ApplicationContextMessageBus now implements the ChannelResolver interface. The MessagePublishingErrorHandler for the SimpleTaskScheduler is now created in the MessageBusParser rather than in the initialization code of the MessageBus.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-20 13:50:22 +00:00
parent 7124894646
commit f8b2724f7a
4 changed files with 28 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
org.springframework.integration/src/main/java/org/springframework/integration/bus/ApplicationContextMessageBus.java
View File

@@ -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;

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

@@ -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(

19
org.springframework.integration/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java
View File

@@ -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

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

@@ -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