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

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.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-11 17:20:52 +00:00
parent f7d04d911b
commit 5a536586e1
6 changed files with 37 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
private final Set<MessageEndpoint> endpoints = new CopyOnWriteArraySet<MessageEndpoint>();
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) {

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

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

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

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

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

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

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

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

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

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