From 2ad8a6ea278bc6a0d53e159c23159a199e5236b0 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 18 Dec 2007 04:32:16 +0000 Subject: [PATCH] Renamed ChannelMapping to ChannelRegistry, and now delegating to DefaultChannelRegistry from the MessageBus. --- ...tionAwareMessagePublishingInterceptor.java | 16 ++--- .../aop/PublisherAnnotationAdvisor.java | 16 ++--- .../integration/bus/MessageBus.java | 43 ++++++------- ...annelMapping.java => ChannelRegistry.java} | 10 ++- ...ngAware.java => ChannelRegistryAware.java} | 6 +- .../channel/DefaultChannelRegistry.java | 62 +++++++++++++++++++ .../config/AnnotationDrivenParser.java | 2 +- ...essageEndpointAnnotationPostProcessor.java | 6 +- .../PublisherAnnotationPostProcessor.java | 22 +++---- .../SubscriberAnnotationPostProcessor.java | 8 +-- .../endpoint/GenericMessageEndpoint.java | 18 +++--- .../integration/endpoint/MessageEndpoint.java | 4 +- .../router/RouterMessageHandlerAdapter.java | 14 ++--- .../aop/PublisherAnnotationAdvisorTests.java | 37 +++-------- .../publisherAnnotationPostProcessorTests.xml | 2 +- 15 files changed, 154 insertions(+), 112 deletions(-) rename spring-integration-core/src/main/java/org/springframework/integration/channel/{ChannelMapping.java => ChannelRegistry.java} (71%) rename spring-integration-core/src/main/java/org/springframework/integration/channel/{ChannelMappingAware.java => ChannelRegistryAware.java} (85%) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java index b00eb024d6..350e4acf4f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java @@ -23,7 +23,7 @@ import org.aopalliance.intercept.MethodInvocation; import org.springframework.aop.support.AopUtils; import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.integration.channel.ChannelMapping; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.util.Assert; @@ -39,17 +39,17 @@ public class AnnotationAwareMessagePublishingInterceptor extends MessagePublishi private String channelAttributeName; - private ChannelMapping channelMapping; + private ChannelRegistry channelRegistry; public AnnotationAwareMessagePublishingInterceptor(Class publisherAnnotationType, - String channelAttributeName, ChannelMapping channelMapping) { - Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null"); - Assert.notNull(channelAttributeName, "channelAttributeName must not be null"); - Assert.notNull(channelMapping, "channelMapping must not be null"); + String channelAttributeName, ChannelRegistry channelRegistry) { + Assert.notNull(publisherAnnotationType, "'publisherAnnotationType' must not be null"); + Assert.notNull(channelAttributeName, "'channelAttributeName' must not be null"); + Assert.notNull(channelRegistry, "'channelRegistry' must not be null"); this.publisherAnnotationType = publisherAnnotationType; this.channelAttributeName = channelAttributeName; - this.channelMapping = channelMapping; + this.channelRegistry = channelRegistry; } @@ -61,7 +61,7 @@ public class AnnotationAwareMessagePublishingInterceptor extends MessagePublishi if (annotation != null) { String channelName = (String) AnnotationUtils.getValue(annotation, this.channelAttributeName); if (channelName != null) { - MessageChannel channel = this.channelMapping.getChannel(channelName); + MessageChannel channel = this.channelRegistry.lookupChannel(channelName); if (channel != null) { return channel; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java index 3b12453573..6913a99309 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java @@ -23,7 +23,7 @@ import org.aopalliance.aop.Advice; import org.springframework.aop.Pointcut; import org.springframework.aop.support.AbstractPointcutAdvisor; import org.springframework.aop.support.annotation.AnnotationMatchingPointcut; -import org.springframework.integration.channel.ChannelMapping; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.util.Assert; /** @@ -42,18 +42,18 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor { private AnnotationMatchingPointcut pointcut; - public PublisherAnnotationAdvisor(ChannelMapping channelMapping) { - this(Publisher.class, "channel", channelMapping); + public PublisherAnnotationAdvisor(ChannelRegistry channelRegistry) { + this(Publisher.class, "channel", channelRegistry); } public PublisherAnnotationAdvisor(Class publisherAnnotationType, String channelNameAttribute, - ChannelMapping channelMapping) { - Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null"); - Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null"); - Assert.notNull(channelMapping, "channelMapping must not be null"); + ChannelRegistry channelRegistry) { + Assert.notNull(publisherAnnotationType, "'publisherAnnotationType' must not be null"); + Assert.notNull(channelNameAttribute, "'channelNameAttribute' must not be null"); + Assert.notNull(channelRegistry, "'channelRegistry' must not be null"); this.pointcut = AnnotationMatchingPointcut.forMethodAnnotation(publisherAnnotationType); this.advice = new AnnotationAwareMessagePublishingInterceptor(publisherAnnotationType, channelNameAttribute, - channelMapping); + channelRegistry); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 44c53c3847..d892822eea 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -31,9 +31,10 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.integration.MessagingException; -import org.springframework.integration.channel.ChannelMapping; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.util.Assert; @@ -44,11 +45,11 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessageBus implements ChannelMapping, ApplicationContextAware, Lifecycle { +public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lifecycle { private Log logger = LogFactory.getLog(this.getClass()); - private Map channels = new ConcurrentHashMap(); + private ChannelRegistry channelRegistry = new DefaultChannelRegistry(); private Map endpoints = new ConcurrentHashMap(); @@ -58,8 +59,6 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life private ScheduledThreadPoolExecutor dispatcherExecutor; - private MessageChannel invalidMessageChannel; - private boolean autoCreateChannels; private boolean running; @@ -78,10 +77,6 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life this.autoCreateChannels = autoCreateChannels; } - public void setInvalidMessageChannel(MessageChannel invalidMessageChannel) { - this.invalidMessageChannel = invalidMessageChannel; - } - @SuppressWarnings("unchecked") private void registerChannels(ApplicationContext context) { Map channelBeans = @@ -117,22 +112,20 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherTasks.size() > 0 ? this.dispatcherTasks.size() : 1); } - public MessageChannel getChannel(String channelName) { - return this.channels.get(channelName); + public MessageChannel getInvalidMessageChannel() { + return this.channelRegistry.getInvalidMessageChannel(); } - public MessageChannel getInvalidMessageChannel() { - return this.invalidMessageChannel; + public void setInvalidMessageChannel(MessageChannel invalidMessageChannel) { + this.channelRegistry.setInvalidMessageChannel(invalidMessageChannel); + } + + public MessageChannel lookupChannel(String channelName) { + return this.channelRegistry.lookupChannel(channelName); } public void registerChannel(String name, MessageChannel channel) { - Assert.notNull(name, "'name' must not be null"); - Assert.notNull(channel, "'channel' must not be null"); - channel.setName(name); - this.channels.put(name, channel); - if (this.logger.isInfoEnabled()) { - logger.info("registered channel '" + name + "'"); - } + this.channelRegistry.registerChannel(name, channel); } public void registerEndpoint(String name, MessageEndpoint endpoint) { @@ -142,7 +135,7 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life if (logger.isInfoEnabled()) { logger.info("registered endpoint '" + name + "'"); } - endpoint.setChannelMapping(this); + endpoint.setChannelRegistry(this); if (endpoint.getInputChannelName() != null && endpoint.getConsumerPolicy() != null) { Subscription subscription = new Subscription(); subscription.setChannel(endpoint.getInputChannelName()); @@ -156,7 +149,7 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life String channelName = subscription.getChannel(); String endpointName = subscription.getEndpoint(); ConsumerPolicy policy = subscription.getPolicy(); - MessageChannel channel = this.channels.get(channelName); + MessageChannel channel = this.lookupChannel(channelName); if (channel == null) { if (this.autoCreateChannels == false) { throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName + @@ -289,7 +282,7 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life return; } for (int i = 0; i < policy.getMaxMessagesPerTask(); i++) { - Message message = channel.receive(this.policy.getReceiveTimeout()); + Message message = channel.receive(this.policy.getReceiveTimeout()); if (message == null) { return; } @@ -342,12 +335,12 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life private MessageEndpoint endpoint; - private Message message; + private Message message; private Throwable error; - EndpointTask(MessageEndpoint endpoint, Message message) { + EndpointTask(MessageEndpoint endpoint, Message message) { this.endpoint = endpoint; this.message = message; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java similarity index 71% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java index a5ed893be3..0217c5a98a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java @@ -17,13 +17,17 @@ package org.springframework.integration.channel; /** - * A strategy interface for resolution of message channels by name. + * A strategy interface for registration and lookup of message channels by name. * * @author Mark Fisher */ -public interface ChannelMapping { +public interface ChannelRegistry { - MessageChannel getChannel(String channelName); + void registerChannel(String name, MessageChannel channel); + + MessageChannel lookupChannel(String channelName); + + void setInvalidMessageChannel(MessageChannel invalidMessageChannel); MessageChannel getInvalidMessageChannel(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMappingAware.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistryAware.java similarity index 85% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMappingAware.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistryAware.java index fefefb091b..267538535b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMappingAware.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistryAware.java @@ -18,12 +18,12 @@ package org.springframework.integration.channel; /** * Interface to be implemented by components that need access to the - * {@link ChannelMapping}. + * {@link ChannelRegistry}. * * @author Mark Fisher */ -public interface ChannelMappingAware { +public interface ChannelRegistryAware { - void setChannelMapping(ChannelMapping channelMapping); + void setChannelRegistry(ChannelRegistry channelRegistry); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java new file mode 100644 index 0000000000..d351f02b1f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java @@ -0,0 +1,62 @@ +/* + * Copyright 2002-2007 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; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.util.Assert; + +/** + * A simple map-backed implementation of {@link ChannelRegistry}. + * + * @author Mark Fisher + */ +public class DefaultChannelRegistry implements ChannelRegistry { + + private Log logger = LogFactory.getLog(this.getClass()); + + private Map channels = new ConcurrentHashMap(); + + private MessageChannel invalidMessageChannel; + + + public void setInvalidMessageChannel(MessageChannel invalidMessageChannel) { + this.invalidMessageChannel = invalidMessageChannel; + } + + public MessageChannel getInvalidMessageChannel() { + return this.invalidMessageChannel; + } + + public MessageChannel lookupChannel(String channelName) { + return this.channels.get(channelName); + } + + public void registerChannel(String name, MessageChannel channel) { + Assert.notNull(name, "'name' must not be null"); + Assert.notNull(channel, "'channel' must not be null"); + channel.setName(name); + this.channels.put(name, channel); + if (logger.isInfoEnabled()) { + logger.info("registered channel '" + name + "'"); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java index f56c9d4915..b6fcc4770f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java @@ -52,7 +52,7 @@ public class AnnotationDrivenParser implements BeanDefinitionParser { private void createPublisherPostProcessor(ParserContext parserContext) { BeanDefinition bd = new RootBeanDefinition(PublisherAnnotationPostProcessor.class); - bd.getPropertyValues().addPropertyValue("channelMapping", + bd.getPropertyValues().addPropertyValue("channelRegistry", new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); BeanComponentDefinition bcd = new BeanComponentDefinition( bd, PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index dab6fbcff2..f6bdc1cb16 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -32,7 +32,7 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.ChannelMappingAware; +import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter; import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; @@ -174,8 +174,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor Annotation annotation = AnnotationUtils.getAnnotation(method, annotationType); if (annotation != null) { MessageHandler handler = handlerCreators.get(annotationType).createHandler(bean, method, annotation); - if (handler instanceof ChannelMappingAware) { - ((ChannelMappingAware) handler).setChannelMapping(messageBus); + if (handler instanceof ChannelRegistryAware) { + ((ChannelRegistryAware) handler).setChannelRegistry(messageBus); } if (handler instanceof InitializingBean) { try { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java index 5a1aded5b8..22fec4d93e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java @@ -27,7 +27,7 @@ import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.integration.aop.Publisher; import org.springframework.integration.aop.PublisherAnnotationAdvisor; -import org.springframework.integration.channel.ChannelMapping; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.util.Assert; /** @@ -42,7 +42,7 @@ public class PublisherAnnotationPostProcessor implements BeanPostProcessor, Bean private String channelNameAttribute = "channel"; - private ChannelMapping channelMapping; + private ChannelRegistry channelRegistry; private Advisor advisor; @@ -50,31 +50,31 @@ public class PublisherAnnotationPostProcessor implements BeanPostProcessor, Bean public void setBeanClassLoader(ClassLoader beanClassLoader) { - Assert.notNull(beanClassLoader, "beanClassLoader must not be null"); + Assert.notNull(beanClassLoader, "'beanClassLoader' must not be null"); this.beanClassLoader = beanClassLoader; } public void setPublisherAnnotationType(Class publisherAnnotationType) { - Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null"); + Assert.notNull(publisherAnnotationType, "'publisherAnnotationType' must not be null"); this.publisherAnnotationType = publisherAnnotationType; } public void setChannelNameAttribute(String channelNameAttribute) { - Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null"); + Assert.notNull(channelNameAttribute, "'channelNameAttribute' must not be null"); this.channelNameAttribute = channelNameAttribute; } - public void setChannelMapping(ChannelMapping channelMapping) { - Assert.notNull(channelMapping, "channelMapping must not be null"); - this.channelMapping = channelMapping; + public void setChannelRegistry(ChannelRegistry channelRegistry) { + Assert.notNull(channelRegistry, "'channelRegistry' must not be null"); + this.channelRegistry = channelRegistry; } private void createAdvisor() { - if (this.channelMapping == null) { - throw new IllegalStateException("channelMapping is required"); + if (this.channelRegistry == null) { + throw new IllegalStateException("'channelRegistry' is required"); } this.advisor = new PublisherAnnotationAdvisor(this.publisherAnnotationType, this.channelNameAttribute, - this.channelMapping); + this.channelRegistry); } public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java index 8ff4d66322..84e1051fb5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java @@ -52,17 +52,17 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { public void setSubscriberAnnotationType(Class subscriberAnnotationType) { - Assert.notNull(subscriberAnnotationType, "subscriberAnnotationType must not be null"); + Assert.notNull(subscriberAnnotationType, "'subscriberAnnotationType' must not be null"); this.subscriberAnnotationType = subscriberAnnotationType; } public void setChannelNameAttribute(String channelNameAttribute) { - Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null"); + Assert.notNull(channelNameAttribute, "'channelNameAttribute' must not be null"); this.channelNameAttribute = channelNameAttribute; } public void setMessageBus(MessageBus messageBus) { - Assert.notNull(messageBus, "messageBus must not be null"); + Assert.notNull(messageBus, "'messageBus' must not be null"); this.messageBus = messageBus; } @@ -87,7 +87,7 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { adapter.afterPropertiesSet(); GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); endpoint.setInputChannelName(channelName); - endpoint.setChannelMapping(messageBus); + endpoint.setChannelRegistry(messageBus); endpoint.setHandler(adapter); String endpointName = ClassUtils.getShortNameAsProperty(targetClass) + "-" + method.getName() + "-endpoint"; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index 8a01fdbf55..2a738fcdef 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -19,7 +19,7 @@ package org.springframework.integration.endpoint; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.channel.ChannelMapping; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; @@ -45,7 +45,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { private MessageHandler handler; - private ChannelMapping channelMapping; + private ChannelRegistry channelRegistry; private ConsumerPolicy consumerPolicy = new ConsumerPolicy(); @@ -87,10 +87,10 @@ public class GenericMessageEndpoint implements MessageEndpoint { } /** - * Set the channel mapping to use for looking up channels by name. + * Set the channel registry to use for looking up channels by name. */ - public void setChannelMapping(ChannelMapping channelMapping) { - this.channelMapping = channelMapping; + public void setChannelRegistry(ChannelRegistry channelRegistry) { + this.channelRegistry = channelRegistry; } @@ -100,7 +100,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { throw new MessagingConfigurationException( "endpoint must have either a 'handler' or 'defaultOutputChannelName'"); } - MessageChannel replyChannel = this.channelMapping.getChannel(this.defaultOutputChannelName); + MessageChannel replyChannel = this.channelRegistry.lookupChannel(this.defaultOutputChannelName); replyChannel.send(message); return; } @@ -117,14 +117,14 @@ public class GenericMessageEndpoint implements MessageEndpoint { } private MessageChannel resolveReplyChannel(Message message) { - if (this.channelMapping == null) { + if (this.channelRegistry == null) { return null; } String replyChannelName = message.getHeader().getReplyChannelName(); if (replyChannelName != null && replyChannelName.trim().length() > 0) { - return this.channelMapping.getChannel(replyChannelName); + return this.channelRegistry.lookupChannel(replyChannelName); } - return this.channelMapping.getChannel(this.defaultOutputChannelName); + return this.channelRegistry.lookupChannel(this.defaultOutputChannelName); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index b2213b43ca..a3afe2dc42 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -17,7 +17,7 @@ package org.springframework.integration.endpoint; import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.channel.ChannelMapping; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.message.Message; /** @@ -35,7 +35,7 @@ public interface MessageEndpoint { ConsumerPolicy getConsumerPolicy(); - void setChannelMapping(ChannelMapping channelMapping); + void setChannelRegistry(ChannelRegistry channelRegistry); void messageReceived(Message message); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java index 5b73dbbc6b..38a8a6845b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java @@ -20,8 +20,8 @@ import java.lang.reflect.Method; import java.util.Collection; import org.springframework.integration.MessagingConfigurationException; -import org.springframework.integration.channel.ChannelMapping; -import org.springframework.integration.channel.ChannelMappingAware; +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.endpoint.SimpleMethodInvoker; import org.springframework.integration.handler.AbstractMessageHandlerAdapter; @@ -34,13 +34,13 @@ import org.springframework.util.StringUtils; * * @author Mark Fisher */ -public class RouterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelMappingAware { +public class RouterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware { private Router routerAnnotation; private Method method; - private ChannelMapping channelMapping; + private ChannelRegistry channelRegistry; public RouterMessageHandlerAdapter(Object object, Method method, Router routerAnnotation) { @@ -49,8 +49,8 @@ public class RouterMessageHandlerAdapter extends AbstractMessageHandlerAdapter i this.routerAnnotation = routerAnnotation; } - public void setChannelMapping(ChannelMapping channelMapping) { - this.channelMapping = channelMapping; + public void setChannelRegistry(ChannelRegistry channelRegistry) { + this.channelRegistry = channelRegistry; } @Override @@ -130,7 +130,7 @@ public class RouterMessageHandlerAdapter extends AbstractMessageHandlerAdapter i } private boolean sendMessage(Message message, String channelName) { - MessageChannel channel = this.channelMapping.getChannel(channelName); + MessageChannel channel = this.channelRegistry.lookupChannel(channelName); if (channel == null) { if (logger.isWarnEnabled()) { logger.warn("unable to resolve channel for name '" + channelName + "'"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java index 3a16bb120e..f54ff20a5b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java @@ -23,9 +23,10 @@ import static org.junit.Assert.assertNull; import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; -import org.springframework.integration.channel.ChannelMapping; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.message.Message; /** @@ -36,21 +37,12 @@ public class PublisherAnnotationAdvisorTests { @Test public void testPublisherAnnotation() { final MessageChannel channel = new PointToPointChannel(); - ChannelMapping channelMapping = new ChannelMapping() { - public MessageChannel getChannel(String channelName) { - if (channelName.equals("testChannel")) { - return channel; - } - return null; - } - public MessageChannel getInvalidMessageChannel() { - return null; - } - }; - PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelMapping); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("testChannel", channel); + PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); proxy.publisherTest(); - Message message = channel.receive(0); + Message message = channel.receive(0); assertNotNull(message); assertEquals("hello world", message.getPayload()); } @@ -58,21 +50,12 @@ public class PublisherAnnotationAdvisorTests { @Test public void testNoPublisherAnnotation() { final MessageChannel channel = new PointToPointChannel(); - ChannelMapping channelMapping = new ChannelMapping() { - public MessageChannel getChannel(String channelName) { - if (channelName.equals("testChannel")) { - return channel; - } - return null; - } - public MessageChannel getInvalidMessageChannel() { - return null; - } - }; - PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelMapping); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("testChannel", channel); + PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); proxy.noPublisherTest(); - Message message = channel.receive(0); + Message message = channel.receive(0); assertNull(message); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml index cbce4ca45e..da37c291b8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml @@ -11,7 +11,7 @@ - +