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

Renamed ChannelMapping to ChannelRegistry, and now delegating to DefaultChannelRegistry from the MessageBus.

Browse Source
This commit is contained in:
Mark Fisher
2007-12-18 04:32:16 +00:00
parent 3edd1e7771
commit 2ad8a6ea27
15 changed files with 154 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-integration-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java
View File

@@ -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<? extends Annotation> 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;
}

16
spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java
View File

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

43
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
private ChannelRegistry channelRegistry = new DefaultChannelRegistry();
private Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
@@ -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<String, MessageChannel> 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;
}

10
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java → spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java
View File

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

6
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelMappingAware.java → spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistryAware.java
View File

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

62
spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java Normal file
View File

@@ -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<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
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 + "'");
}
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java
View File

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

6
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

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

22
spring-integration-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java
View File

@@ -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<? extends Annotation> 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 {

8
spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java
View File

@@ -52,17 +52,17 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
public void setSubscriberAnnotationType(Class<? extends Annotation> 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";

18
spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java
View File

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

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

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

14
spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java
View File

@@ -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 + "'");