diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java index 36ff9c7fd4..4e00e5cd17 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java @@ -19,7 +19,6 @@ package org.springframework.integration.bus; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.RendezvousChannel; import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; @@ -37,7 +36,7 @@ public class DefaultErrorChannel extends RendezvousChannel { public DefaultErrorChannel() { this.addInterceptor(new ErrorLoggingInterceptor()); - this.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME); + this.setBeanName(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME); } 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 32526674e5..f399f74859 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 @@ -35,7 +35,6 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationListener; import org.springframework.context.Lifecycle; import org.springframework.context.event.ContextRefreshedEvent; -import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.endpoint.MessageEndpoint; @@ -54,6 +53,9 @@ import org.springframework.util.Assert; */ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, ApplicationListener, DisposableBean { + public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel"; + + private final Log logger = LogFactory.getLog(this.getClass()); private final Set endpoints = new CopyOnWriteArraySet(); @@ -233,7 +235,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A } if (this.taskScheduler instanceof SimpleTaskScheduler) { ((SimpleTaskScheduler) this.taskScheduler).setErrorHandler( - new MessagePublishingErrorHandler(this.lookupChannel(ChannelRegistry.ERROR_CHANNEL_NAME))); + new MessagePublishingErrorHandler(this.lookupChannel(ERROR_CHANNEL_BEAN_NAME))); } this.taskScheduler.start(); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java index 5ac77995ba..314828077b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -17,7 +17,6 @@ package org.springframework.integration.bus; import org.springframework.context.Lifecycle; -import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.endpoint.MessageEndpoint; /** @@ -25,7 +24,7 @@ import org.springframework.integration.endpoint.MessageEndpoint; * * @author Mark Fisher */ -public interface MessageBus extends ChannelRegistry, Lifecycle { +public interface MessageBus extends Lifecycle { void registerEndpoint(MessageEndpoint endpoint); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java deleted file mode 100644 index b021c69871..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelRegistry.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * Copyright 2002-2008 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.channel; - -/** - * A strategy interface for registration and lookup of message channels by name. - * - * @author Mark Fisher - */ -public interface ChannelRegistry { - - static final String ERROR_CHANNEL_NAME = "errorChannel"; - - MessageChannel lookupChannel(String channelName); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java index c3784eb098..0e31d07b07 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java @@ -22,12 +22,13 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.Poller; -import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.BeanFactoryChannelResolver; +import org.springframework.integration.channel.ChannelResolutionException; +import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; -import org.springframework.integration.config.xml.MessageBusParser; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.PollingConsumerEndpoint; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; @@ -48,28 +49,20 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo private final ConfigurableBeanFactory beanFactory; - private final ChannelRegistry channelRegistry; + private final ChannelResolver channelResolver; public ChannelAdapterAnnotationPostProcessor(ConfigurableBeanFactory beanFactory) { Assert.notNull(beanFactory, "BeanFactory must not be null"); this.beanFactory = beanFactory; - this.channelRegistry = (ChannelRegistry) - this.beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); + this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory); } public Object postProcess(Object bean, String beanName, Method method, ChannelAdapter annotation) { Assert.notNull(this.beanFactory, "BeanFactory must not be null"); MessageEndpoint endpoint = null; - String channelName = annotation.value(); - MessageChannel channel = this.channelRegistry.lookupChannel(channelName); - if (channel == null) { - DirectChannel directChannel = new DirectChannel(); - directChannel.setBeanName(channelName); - this.beanFactory.registerSingleton(channelName, directChannel); - channel = directChannel; - } + MessageChannel channel = this.resolveOrCreateChannel(annotation.value()); Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class); if (method.getParameterTypes().length == 0 && hasReturnValue(method)) { MethodInvokingSource source = new MethodInvokingSource(); @@ -94,6 +87,18 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo return bean; } + private MessageChannel resolveOrCreateChannel(String channelName) { + try { + return this.channelResolver.resolveChannelName(channelName); + } + catch (ChannelResolutionException e) { + DirectChannel directChannel = new DirectChannel(); + directChannel.setBeanName(channelName); + this.beanFactory.registerSingleton(channelName, directChannel); + return directChannel; + } + } + private SourcePollingChannelAdapter createInboundChannelAdapter(MethodInvokingSource source, MessageChannel channel, Poller pollerAnnotation) { Assert.notNull(pollerAnnotation, "The @Poller annotation is required (at method-level) " + "when using the @ChannelAdapter annotation with a no-arg method."); 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 a57df878b7..886bc52a30 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,7 +41,6 @@ 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; @@ -101,10 +100,10 @@ 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)) { + if (!parserContext.getRegistry().containsBeanDefinition(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME)) { RootBeanDefinition errorChannelDef = new RootBeanDefinition(QueueChannel.class); BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder( - errorChannelDef, ChannelRegistry.ERROR_CHANNEL_NAME); + errorChannelDef, DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME); BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, parserContext.getRegistry()); } if (StringUtils.hasText(taskSchedulerRef)) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 8698ad4cf3..01d0a47c4f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -34,6 +34,8 @@ import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.BeanFactoryChannelResolver; +import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.config.xml.MessageBusParser; @@ -72,6 +74,8 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, private volatile MessageBus messageBus; + private ChannelResolver channelResolver; + private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -137,6 +141,7 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, public void setBeanFactory(BeanFactory beanFactory) { this.messageBus = (MessageBus) beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); } public void afterPropertiesSet() throws Exception { @@ -225,12 +230,12 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, Assert.state(this.messageBus != null, "MessageBus is required for channel resolution"); String requestChannelName = gatewayAnnotation.requestChannel(); if (StringUtils.hasText(requestChannelName)) { - requestChannel = this.messageBus.lookupChannel(requestChannelName); + requestChannel = this.channelResolver.resolveChannelName(requestChannelName); Assert.notNull(requestChannel, "failed to resolve request channel '" + requestChannelName + "'"); } String replyChannelName = gatewayAnnotation.replyChannel(); if (StringUtils.hasText(replyChannelName)) { - replyChannel = this.messageBus.lookupChannel(replyChannelName); + replyChannel = this.channelResolver.resolveChannelName(replyChannelName); Assert.notNull(replyChannel, "failed to resolve reply channel '" + replyChannelName + "'"); } requestTimeout = gatewayAnnotation.requestTimeout(); 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 028a2f32ba..b3be17e722 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 @@ -30,7 +30,6 @@ 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.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.PublishSubscribeChannel; @@ -243,8 +242,8 @@ public class DefaultMessageBusTests { public void consumerSubscribedToErrorChannel() throws InterruptedException { GenericApplicationContext context = new GenericApplicationContext(); QueueChannel errorChannel = new QueueChannel(); - errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME); - context.getBeanFactory().registerSingleton(ChannelRegistry.ERROR_CHANNEL_NAME, errorChannel); + errorChannel.setBeanName(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME); + context.getBeanFactory().registerSingleton(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME, errorChannel); final CountDownLatch latch = new CountDownLatch(1); AbstractReplyProducingMessageConsumer consumer = new AbstractReplyProducingMessageConsumer() { public Message handle(Message message) { 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 293dd0d572..4ad235134a 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 @@ -24,7 +24,6 @@ import org.junit.Test; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.ThreadLocalChannel; @@ -110,8 +109,9 @@ public class DirectChannelSubscriptionTests { @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); + errorChannel.setBeanName(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME); + context.getBeanFactory().registerSingleton( + DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME, errorChannel); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java index fcec258c30..f119c7254f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java @@ -24,6 +24,8 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.BeanFactoryChannelResolver; +import org.springframework.integration.channel.ChannelResolutionException; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; @@ -48,7 +50,8 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests assertTrue(channel instanceof DirectChannel); MessageBus bus = (MessageBus) this.applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); bus.start(); - assertNotNull(bus.lookupChannel(beanName)); + BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext); + assertNotNull(channelResolver.resolveChannelName(beanName)); Object adapter = this.applicationContext.getBean(beanName + ".adapter"); assertNotNull(adapter); assertTrue(adapter instanceof SubscribingConsumerEndpoint); @@ -68,7 +71,8 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests assertTrue(channel instanceof DirectChannel); MessageBus bus = (MessageBus) this.applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); bus.start(); - assertNotNull(bus.lookupChannel(beanName)); + BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext); + assertNotNull(channelResolver.resolveChannelName(beanName)); Object adapter = this.applicationContext.getBean(beanName + ".adapter"); assertNotNull(adapter); assertTrue(adapter instanceof SubscribingConsumerEndpoint); @@ -87,7 +91,6 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel"); MessageBus bus = (MessageBus) this.applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); bus.start(); - assertNull(bus.lookupChannel(beanName)); Object adapter = this.applicationContext.getBean(beanName); assertNotNull(adapter); assertTrue(adapter instanceof SourcePollingChannelAdapter); @@ -99,4 +102,10 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests bus.stop(); } + @Test(expected = ChannelResolutionException.class) + public void methodInvokingSourceAdapterIsNotChannel() { + BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext); + channelResolver.resolveChannelName("methodInvokingSource"); + } + } 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 4f9efe0e80..c61451bc6e 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 @@ -36,7 +36,6 @@ import org.springframework.integration.bus.MessageBusInterceptorTests; import org.springframework.integration.bus.TestMessageBusAwareImpl; import org.springframework.integration.bus.TestMessageBusStartInterceptor; import org.springframework.integration.bus.TestMessageBusStopInterceptor; -import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.config.xml.MessageBusParser; import org.springframework.integration.scheduling.TaskScheduler; @@ -54,7 +53,7 @@ public class MessageBusParserTests { "messageBusWithErrorChannel.xml", this.getClass()); DefaultMessageBus bus = (DefaultMessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); bus.initialize(); - MessageChannel channel = bus.lookupChannel(ChannelRegistry.ERROR_CHANNEL_NAME); + MessageChannel channel = bus.lookupChannel(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME); assertEquals(context.getBean("errorChannel"), channel); } @@ -65,7 +64,7 @@ public class MessageBusParserTests { DefaultMessageBus bus = (DefaultMessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); bus.initialize(); assertNotNull("parser should have created a default error channel", - bus.lookupChannel(ChannelRegistry.ERROR_CHANNEL_NAME)); + bus.lookupChannel(DefaultMessageBus.ERROR_CHANNEL_BEAN_NAME)); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index 473e42cdd3..0486e1d13f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -34,8 +34,8 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.aggregator.AbstractMessageAggregator; import org.springframework.integration.aggregator.CompletionStrategyAdapter; import org.springframework.integration.aggregator.SequenceSizeCompletionStrategy; -import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.config.xml.MessageBusParser; +import org.springframework.integration.channel.BeanFactoryChannelResolver; +import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.endpoint.SubscribingConsumerEndpoint; /** @@ -71,9 +71,10 @@ public class AggregatorAnnotationTests { AbstractMessageAggregator aggregator = this.getAggregator(context, endpointName); assertTrue(getPropertyValue(aggregator, "completionStrategy") instanceof SequenceSizeCompletionStrategy); - assertEquals(getMessageBus(context).lookupChannel("outputChannel"), + ChannelResolver channelResolver = new BeanFactoryChannelResolver(context); + assertEquals(channelResolver.resolveChannelName("outputChannel"), getPropertyValue(aggregator, "outputChannel")); - assertEquals(getMessageBus(context).lookupChannel("discardChannel"), + assertEquals(channelResolver.resolveChannelName("discardChannel"), getPropertyValue(aggregator, "discardChannel")); assertEquals(98765432l, getPropertyValue(aggregator, "channelTemplate.sendTimeout")); assertEquals(4567890l, getPropertyValue(aggregator, "timeout")); @@ -107,8 +108,4 @@ public class AggregatorAnnotationTests { return (AbstractMessageAggregator) new DirectFieldAccessor(endpoint).getPropertyValue("consumer"); } - private MessageBus getMessageBus(ApplicationContext context) { - return (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); - } - } 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 a0971d67dc..8179e21ba4 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 @@ -38,6 +38,8 @@ import org.springframework.integration.annotation.Poller; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.bus.DefaultMessageBus; +import org.springframework.integration.channel.BeanFactoryChannelResolver; +import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; @@ -152,7 +154,8 @@ public class MessagingAnnotationPostProcessorTests { TargetAnnotationTestBean testBean = new TargetAnnotationTestBean(latch); postProcessor.postProcessAfterInitialization(testBean, "testBean"); messageBus.start(); - MessageChannel testChannel = messageBus.lookupChannel("testChannel"); + ChannelResolver channelResolver = new BeanFactoryChannelResolver(context); + MessageChannel testChannel = channelResolver.resolveChannelName("testChannel"); testChannel.send(new StringMessage("foo")); latch.await(1000, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); @@ -397,7 +400,8 @@ public class MessagingAnnotationPostProcessorTests { ChannelAdapterAnnotationTestBean testBean = new ChannelAdapterAnnotationTestBean(); postProcessor.postProcessAfterInitialization(testBean, "testBean"); messageBus.start(); - DirectChannel testChannel = (DirectChannel) messageBus.lookupChannel("testChannel"); + ChannelResolver channelResolver = new BeanFactoryChannelResolver(context); + DirectChannel testChannel = (DirectChannel) channelResolver.resolveChannelName("testChannel"); final CountDownLatch latch = new CountDownLatch(1); final AtomicReference> receivedMessage = new AtomicReference>(); testChannel.subscribe(new MessageConsumer() {