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

Removed EndpointRegistry. DefaultMessageBus now delegates to the ApplicationContext for all endpoint lookups, and the annotation-based post-processing registers singletons with the context rather than going through the DefaultMessageBus.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-18 22:42:08 +00:00
parent fe14643a8d
commit eb5884a7d6
18 changed files with 226 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,10 +16,10 @@
package org.springframework.integration.bus;
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;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -30,6 +30,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
@@ -69,8 +70,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
private final ChannelRegistry channelRegistry = new DefaultChannelRegistry();
private final Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
private final MessageBusInterceptorsList interceptors = new MessageBusInterceptorsList();
private final Set<Lifecycle> lifecycleGateways = new CopyOnWriteArraySet<Lifecycle>();
@@ -171,6 +170,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
return;
}
this.initializing = true;
Assert.notNull(this.applicationContext, "ApplicationContext must not be null");
if (this.taskScheduler == null) {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE);
executor.setThreadFactory(new CustomizableThreadFactory("message-bus-"));
@@ -214,8 +214,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
public void registerEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
Assert.notNull(endpoint.getName(), "endpoint name must not be null");
this.endpoints.put(endpoint.getName(), endpoint);
if (this.isRunning()) {
this.activateEndpoint(endpoint);
}
@@ -224,26 +222,27 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
}
}
public MessageEndpoint unregisterEndpoint(String name) {
Assert.notNull(name, "endpoint name must not be null");
MessageEndpoint endpoint = this.endpoints.remove(name);
if (endpoint == null) {
return null;
}
this.deactivateEndpoint(endpoint);
return endpoint;
}
public MessageEndpoint lookupEndpoint(String endpointName) {
return this.endpoints.get(endpointName);
if (this.applicationContext.containsBean(endpointName)) {
Object bean = this.applicationContext.getBean(endpointName);
if (bean instanceof MessageEndpoint) {
return (MessageEndpoint) bean;
}
}
return null;
}
public Set<String> getEndpointNames() {
return this.endpoints.keySet();
public String[] getEndpointNames() {
return this.applicationContext.getBeanNamesForType(MessageEndpoint.class);
}
private Collection<MessageEndpoint> getEndpoints() {
GenericBeanFactoryAccessor accessor = new GenericBeanFactoryAccessor(this.applicationContext);
return accessor.getBeansOfType(MessageEndpoint.class).values();
}
private void activateEndpoints() {
for (MessageEndpoint endpoint : this.endpoints.values()) {
for (MessageEndpoint endpoint : this.getEndpoints()) {
if (endpoint != null) {
this.activateEndpoint(endpoint);
}
@@ -251,7 +250,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
}
private void deactivateEndpoints() {
for (MessageEndpoint endpoint : this.endpoints.values()) {
for (MessageEndpoint endpoint : this.getEndpoints()) {
if (endpoint != null) {
this.deactivateEndpoint(endpoint);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -19,15 +19,17 @@ package org.springframework.integration.bus;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.MessageEndpoint;
/**
* The message bus interface.
*
* @author Mark Fisher
*/
public interface MessageBus extends ChannelRegistry, EndpointRegistry, Lifecycle {
public interface MessageBus extends ChannelRegistry, Lifecycle {
MessageChannel getErrorChannel();
void registerEndpoint(MessageEndpoint endpoint);
}

15
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -31,7 +31,6 @@ import org.springframework.integration.endpoint.AbstractInOutEndpoint;
import org.springframework.integration.endpoint.AbstractMessageConsumingEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
/**
@@ -66,9 +65,6 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
if (endpoint != null) {
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
this.configureEndpoint(endpoint, annotation, pollerAnnotation);
if (endpoint.getName() == null) {
endpoint.setBeanName(this.generateEndpointName(beanName, annotation));
}
endpoint.afterPropertiesSet();
return endpoint;
}
@@ -117,17 +113,6 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
}
private String generateEndpointName(String beanName, T annotation) {
String endpointName = beanName + "." + ClassUtils.getShortNameAsProperty(annotation.annotationType());
String id = endpointName;
int counter = 0;
while (this.messageBus.lookupEndpoint(id) != null) {
id = endpointName + "#" + counter;
counter++;
}
return id;
}
protected abstract Object createMethodInvokingAdapter(Object bean, Method method, T annotation);
protected abstract AbstractEndpoint createEndpoint(Object originalBean, Object adapter);

29
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.config.annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.ChannelAdapter;
@@ -26,14 +27,15 @@ import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.OutboundChannelAdapter;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.handler.MethodInvokingTarget;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* Post-processor for methods annotated with {@link ChannelAdapter @ChannelAdapter}.
@@ -44,21 +46,26 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
private final MessageBus messageBus;
private final ConfigurableBeanFactory beanFactory;
public ChannelAdapterAnnotationPostProcessor(MessageBus messageBus) {
public ChannelAdapterAnnotationPostProcessor(MessageBus messageBus, ConfigurableBeanFactory beanFactory) {
Assert.notNull(messageBus, "MessageBus must not be null");
Assert.notNull(beanFactory, "BeanFactory must not be null");
this.messageBus = messageBus;
this.beanFactory = 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.messageBus.lookupChannel(channelName);
if (channel == null) {
DirectChannel directChannel = new DirectChannel();
directChannel.setBeanName(channelName);
this.messageBus.registerChannel(directChannel);
this.beanFactory.registerSingleton(channelName, directChannel);
channel = directChannel;
}
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
@@ -77,7 +84,9 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
+ " a return value (inbound) or methods that have no return value but do accept arguments (outbound)");
}
if (endpoint != null) {
this.messageBus.registerEndpoint(endpoint);
String annotationName = ClassUtils.getShortNameAsProperty(annotation.annotationType());
String endpointName = beanName + "." + method.getName() + "." + annotationName;
this.beanFactory.registerSingleton(endpointName, endpoint);
}
return bean;
}
@@ -92,7 +101,6 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
adapter.setSource(source);
adapter.setOutputChannel(channel);
adapter.setSchedule(schedule);
adapter.setBeanName(this.generateUniqueName(channel.getName() + ".inboundAdapter"));
return adapter;
}
@@ -105,7 +113,6 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
: new PollingSchedule(0);
adapter.setSchedule(schedule);
}
adapter.setBeanName(this.generateUniqueName(channel.getName() + ".outboundAdapter"));
return adapter;
}
@@ -121,14 +128,4 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
return !method.getReturnType().equals(void.class);
}
private String generateUniqueName(String name) {
int counter = 0;
String id = name;
while (this.messageBus.lookupEndpoint(id) != null) {
id = name + "#" + counter;
counter++;
}
return id;
}
}

34
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -30,8 +30,12 @@ import org.springframework.aop.support.AopUtils;
import org.springframework.aop.support.DelegatingIntroductionInterceptor;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Aggregator;
@@ -55,10 +59,12 @@ import org.springframework.util.ReflectionUtils;
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, InitializingBean, BeanClassLoaderAware {
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean, BeanClassLoaderAware {
private final MessageBus messageBus;
private volatile ConfigurableBeanFactory beanFactory;
private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors =
@@ -71,13 +77,20 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
}
public void setBeanFactory(BeanFactory beanFactory) {
Assert.isAssignable(ConfigurableBeanFactory.class, beanFactory.getClass(),
"a ConfigurableBeanFactory is required");
this.beanFactory = (ConfigurableBeanFactory) beanFactory;
}
public void setBeanClassLoader(ClassLoader beanClassLoader) {
this.beanClassLoader = beanClassLoader;
}
public void afterPropertiesSet() {
Assert.notNull(this.beanFactory, "BeanFactory must not be null");
postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.messageBus));
postProcessors.put(ChannelAdapter.class, new ChannelAdapterAnnotationPostProcessor(this.messageBus));
postProcessors.put(ChannelAdapter.class, new ChannelAdapterAnnotationPostProcessor(this.messageBus, this.beanFactory));
postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.messageBus));
postProcessors.put(ServiceActivator.class, new ServiceActivatorAnnotationPostProcessor(this.messageBus));
postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.messageBus));
@@ -89,6 +102,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
}
public Object postProcessAfterInitialization(Object bean, final String beanName) throws BeansException {
Assert.notNull(this.beanFactory, "BeanFactory must not be null");
final Object originalBean = bean;
final Class<?> beanClass = this.getBeanClass(bean);
if (!this.isStereotype(beanClass)) {
@@ -107,7 +121,11 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
Object result = postProcessor.postProcess(originalBean, beanName, method, annotation);
if (result != null) {
if (result instanceof MessageEndpoint) {
messageBus.registerEndpoint((MessageEndpoint) result);
String endpointBeanName = generateBeanName(beanName, method, annotation.annotationType());
if (result instanceof BeanNameAware) {
((BeanNameAware) result).setBeanName(endpointBeanName);
}
beanFactory.registerSingleton(endpointBeanName, result);
}
else {
boolean shouldProxy = false;
@@ -163,4 +181,14 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
return false;
}
private String generateBeanName(String originalBeanName, Method method, Class<? extends Annotation> annotationType) {
String baseName = originalBeanName + "." + method.getName() + "." + ClassUtils.getShortNameAsProperty(annotationType);
String name = baseName;
int count = 1;
while (this.beanFactory.containsBean(name)) {
name = baseName + "#" + (++count);
}
return name;
}
}

36
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java
View File

@@ -1,36 +0,0 @@
/*
* 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.endpoint;
import java.util.Set;
/**
* A strategy interface for registration and lookup of message endpoints by name.
*
* @author Mark Fisher
*/
public interface EndpointRegistry {
void registerEndpoint(MessageEndpoint endpoint);
MessageEndpoint unregisterEndpoint(String name);
MessageEndpoint lookupEndpoint(String endpointName);
Set<String> getEndpointNames();
}

2
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -24,6 +24,4 @@ package org.springframework.integration.endpoint;
*/
public interface MessageEndpoint {
String getName();
}

11
org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java
View File

@@ -21,7 +21,6 @@ import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.bus.MessageBusAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.handler.ReplyMessageCorrelator;
import org.springframework.integration.message.DefaultMessageCreator;
@@ -53,7 +52,7 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M
private volatile ReplyMessageCorrelator replyMessageCorrelator;
private volatile EndpointRegistry endpointRegistry;
private volatile MessageBus messageBus;
private final Object replyMessageCorrelatorMonitor = new Object();
@@ -97,7 +96,7 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M
}
public void setMessageBus(MessageBus messageBus) {
this.endpointRegistry = messageBus;
this.messageBus = messageBus;
}
public void send(Object object) {
@@ -160,14 +159,14 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M
if (this.replyMessageCorrelator != null) {
return;
}
if (this.endpointRegistry == null) {
throw new ConfigurationException("No EndpointRegistry available. Cannot register ReplyMessageCorrelator.");
if (this.messageBus == null) {
throw new ConfigurationException("No MessageBus available. Cannot register ReplyMessageCorrelator.");
}
ReplyMessageCorrelator correlator = new ReplyMessageCorrelator();
correlator.setBeanName("internal.correlator." + this);
correlator.setInputChannel(this.replyChannel);
correlator.afterPropertiesSet();
this.endpointRegistry.registerEndpoint(correlator);
this.messageBus.registerEndpoint(correlator);
this.replyMessageCorrelator = correlator;
}
}