Publisher annotation interceptor now delegates to a ChannelResolver instance rather than a ChannelRegistry.
This commit is contained in:
@@ -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.ChannelRegistry;
|
||||
import org.springframework.integration.channel.ChannelResolver;
|
||||
import org.springframework.integration.channel.MessageChannel;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
@@ -39,17 +39,17 @@ public class AnnotationAwareMessagePublishingInterceptor extends MessagePublishi
|
||||
|
||||
private String channelAttributeName;
|
||||
|
||||
private ChannelRegistry channelRegistry;
|
||||
private ChannelResolver channelResolver;
|
||||
|
||||
|
||||
public AnnotationAwareMessagePublishingInterceptor(Class<? extends Annotation> publisherAnnotationType,
|
||||
String channelAttributeName, ChannelRegistry channelRegistry) {
|
||||
String channelAttributeName, ChannelResolver channelResolver) {
|
||||
Assert.notNull(publisherAnnotationType, "'publisherAnnotationType' must not be null");
|
||||
Assert.notNull(channelAttributeName, "'channelAttributeName' must not be null");
|
||||
Assert.notNull(channelRegistry, "'channelRegistry' must not be null");
|
||||
Assert.notNull(channelResolver, "'channelResolver' must not be null");
|
||||
this.publisherAnnotationType = publisherAnnotationType;
|
||||
this.channelAttributeName = channelAttributeName;
|
||||
this.channelRegistry = channelRegistry;
|
||||
this.channelResolver = channelResolver;
|
||||
}
|
||||
|
||||
|
||||
@@ -57,7 +57,7 @@ public class AnnotationAwareMessagePublishingInterceptor extends MessagePublishi
|
||||
protected MessageChannel resolveChannel(MethodInvocation invocation) {
|
||||
String channelName = this.extractAnnotationValue(invocation, this.channelAttributeName, String.class);
|
||||
if (channelName != null) {
|
||||
MessageChannel channel = this.channelRegistry.lookupChannel(channelName);
|
||||
MessageChannel channel = this.channelResolver.resolveChannelName(channelName);
|
||||
if (channel != null) {
|
||||
return channel;
|
||||
}
|
||||
|
||||
@@ -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.ChannelRegistry;
|
||||
import org.springframework.integration.channel.ChannelResolver;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
@@ -42,18 +42,18 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor {
|
||||
private AnnotationMatchingPointcut pointcut;
|
||||
|
||||
|
||||
public PublisherAnnotationAdvisor(ChannelRegistry channelRegistry) {
|
||||
this(Publisher.class, "channel", channelRegistry);
|
||||
public PublisherAnnotationAdvisor(ChannelResolver channelResolver) {
|
||||
this(Publisher.class, "channel", channelResolver);
|
||||
}
|
||||
|
||||
public PublisherAnnotationAdvisor(Class<? extends Annotation> publisherAnnotationType, String channelNameAttribute,
|
||||
ChannelRegistry channelRegistry) {
|
||||
public PublisherAnnotationAdvisor(Class<? extends Annotation> publisherAnnotationType,
|
||||
String channelNameAttribute, ChannelResolver channelResolver) {
|
||||
Assert.notNull(publisherAnnotationType, "'publisherAnnotationType' must not be null");
|
||||
Assert.notNull(channelNameAttribute, "'channelNameAttribute' must not be null");
|
||||
Assert.notNull(channelRegistry, "'channelRegistry' must not be null");
|
||||
Assert.notNull(channelResolver, "'channelResolver' must not be null");
|
||||
this.pointcut = AnnotationMatchingPointcut.forMethodAnnotation(publisherAnnotationType);
|
||||
this.advice = new AnnotationAwareMessagePublishingInterceptor(publisherAnnotationType, channelNameAttribute,
|
||||
channelRegistry);
|
||||
this.advice = new AnnotationAwareMessagePublishingInterceptor(
|
||||
publisherAnnotationType, channelNameAttribute, channelResolver);
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -26,11 +26,13 @@ import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.aop.support.AopUtils;
|
||||
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.config.BeanPostProcessor;
|
||||
import org.springframework.core.annotation.AnnotationUtils;
|
||||
import org.springframework.integration.aop.Publisher;
|
||||
import org.springframework.integration.aop.PublisherAnnotationAdvisor;
|
||||
import org.springframework.integration.channel.ChannelRegistry;
|
||||
import org.springframework.integration.channel.BeanFactoryChannelResolver;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
|
||||
@@ -40,19 +42,23 @@ import org.springframework.util.ReflectionUtils;
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class PublisherAnnotationPostProcessor implements BeanPostProcessor, BeanClassLoaderAware {
|
||||
public class PublisherAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, BeanClassLoaderAware {
|
||||
|
||||
private volatile Class<? extends Annotation> publisherAnnotationType = Publisher.class;
|
||||
|
||||
private volatile String channelNameAttribute = "channel";
|
||||
|
||||
private ChannelRegistry channelRegistry;
|
||||
private volatile Advisor advisor;
|
||||
|
||||
private Advisor advisor;
|
||||
private volatile BeanFactory beanFactory;
|
||||
|
||||
private ClassLoader beanClassLoader;
|
||||
private volatile ClassLoader beanClassLoader;
|
||||
|
||||
|
||||
public void setBeanFactory(BeanFactory beanFactory) {
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
public void setBeanClassLoader(ClassLoader beanClassLoader) {
|
||||
Assert.notNull(beanClassLoader, "'beanClassLoader' must not be null");
|
||||
this.beanClassLoader = beanClassLoader;
|
||||
@@ -68,17 +74,10 @@ public class PublisherAnnotationPostProcessor implements BeanPostProcessor, Bean
|
||||
this.channelNameAttribute = channelNameAttribute;
|
||||
}
|
||||
|
||||
public void setChannelRegistry(ChannelRegistry channelRegistry) {
|
||||
Assert.notNull(channelRegistry, "'channelRegistry' must not be null");
|
||||
this.channelRegistry = channelRegistry;
|
||||
}
|
||||
|
||||
private void createAdvisor() {
|
||||
if (this.channelRegistry == null) {
|
||||
throw new IllegalStateException("'channelRegistry' is required");
|
||||
}
|
||||
Assert.state(this.beanFactory != null, "BeanFactory is required");
|
||||
this.advisor = new PublisherAnnotationAdvisor(this.publisherAnnotationType, this.channelNameAttribute,
|
||||
this.channelRegistry);
|
||||
new BeanFactoryChannelResolver(this.beanFactory));
|
||||
}
|
||||
|
||||
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
|
||||
|
||||
@@ -178,10 +178,7 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
|
||||
|
||||
private void registerPublisherPostProcessor(ParserContext parserContext) {
|
||||
BeanDefinition bd = new RootBeanDefinition(PublisherAnnotationPostProcessor.class);
|
||||
bd.getPropertyValues().addPropertyValue("channelRegistry",
|
||||
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
|
||||
BeanComponentDefinition bcd = new BeanComponentDefinition(
|
||||
bd, PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
|
||||
BeanComponentDefinition bcd = new BeanComponentDefinition(bd, PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
|
||||
parserContext.registerBeanComponent(bcd);
|
||||
}
|
||||
|
||||
|
||||
@@ -21,15 +21,11 @@ import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.integration.channel.ChannelRegistry;
|
||||
import org.springframework.integration.channel.MessageChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.channel.TestChannelResolver;
|
||||
import org.springframework.integration.message.Message;
|
||||
|
||||
/**
|
||||
@@ -41,9 +37,9 @@ public class PublisherAnnotationAdvisorTests {
|
||||
public void testPublisherAnnotation() {
|
||||
final QueueChannel channel = new QueueChannel();
|
||||
channel.setBeanName("testChannel");
|
||||
TestChannelRegistry channelRegistry = new TestChannelRegistry();
|
||||
channelRegistry.registerChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry);
|
||||
TestChannelResolver channelResolver = new TestChannelResolver();
|
||||
channelResolver.addChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelResolver);
|
||||
TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor);
|
||||
proxy.publisherTest();
|
||||
Message<?> message = channel.receive(0);
|
||||
@@ -55,9 +51,9 @@ public class PublisherAnnotationAdvisorTests {
|
||||
public void testNoPublisherAnnotation() {
|
||||
final QueueChannel channel = new QueueChannel();
|
||||
channel.setBeanName("testChannel");
|
||||
TestChannelRegistry channelRegistry = new TestChannelRegistry();
|
||||
channelRegistry.registerChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry);
|
||||
TestChannelResolver channelResolver = new TestChannelResolver();
|
||||
channelResolver.addChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelResolver);
|
||||
TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor);
|
||||
proxy.noPublisherTest();
|
||||
Message<?> message = channel.receive(0);
|
||||
@@ -68,9 +64,9 @@ public class PublisherAnnotationAdvisorTests {
|
||||
public void testPublishArguments() {
|
||||
final QueueChannel channel = new QueueChannel();
|
||||
channel.setBeanName("testChannel");
|
||||
TestChannelRegistry channelRegistry = new TestChannelRegistry();
|
||||
channelRegistry.registerChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry);
|
||||
TestChannelResolver channelResolver = new TestChannelResolver();
|
||||
channelResolver.addChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelResolver);
|
||||
TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor);
|
||||
proxy.publishArguments("foo", 99);
|
||||
Message<?> message = channel.receive(0);
|
||||
@@ -86,9 +82,9 @@ public class PublisherAnnotationAdvisorTests {
|
||||
public void testPublishException() {
|
||||
final QueueChannel channel = new QueueChannel();
|
||||
channel.setBeanName("testChannel");
|
||||
TestChannelRegistry channelRegistry = new TestChannelRegistry();
|
||||
channelRegistry.registerChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry);
|
||||
TestChannelResolver channelResolver = new TestChannelResolver();
|
||||
channelResolver.addChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelResolver);
|
||||
TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor);
|
||||
RuntimeException caughtException = null;
|
||||
try {
|
||||
@@ -109,9 +105,9 @@ public class PublisherAnnotationAdvisorTests {
|
||||
public void testPublishReturnValue() {
|
||||
final QueueChannel channel = new QueueChannel();
|
||||
channel.setBeanName("testChannel");
|
||||
TestChannelRegistry channelRegistry = new TestChannelRegistry();
|
||||
channelRegistry.registerChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry);
|
||||
TestChannelResolver channelResolver = new TestChannelResolver();
|
||||
channelResolver.addChannel(channel);
|
||||
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelResolver);
|
||||
TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor);
|
||||
Integer actualReturnValue = proxy.publishReturnValue();
|
||||
Message<?> message = channel.receive(0);
|
||||
@@ -174,18 +170,4 @@ public class PublisherAnnotationAdvisorTests {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class TestChannelRegistry implements ChannelRegistry {
|
||||
|
||||
private final Map<String, MessageChannel> channels = new HashMap<String, MessageChannel>();
|
||||
|
||||
public MessageChannel lookupChannel(String channelName) {
|
||||
return this.channels.get(channelName);
|
||||
}
|
||||
|
||||
public void registerChannel(MessageChannel channel) {
|
||||
this.channels.put(channel.getName(), channel);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,24 +1,13 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans
|
||||
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
|
||||
|
||||
<bean id="messageBus" class="org.springframework.integration.bus.DefaultMessageBus">
|
||||
<property name="taskScheduler">
|
||||
<bean class="org.springframework.integration.util.TestUtils"
|
||||
factory-method="createTaskScheduler">
|
||||
<constructor-arg value="10"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
|
||||
|
||||
<bean id="testChannel" class="org.springframework.integration.channel.QueueChannel"/>
|
||||
|
||||
<bean id="testBean" class="org.springframework.integration.aop.PublisherAnnotationTestBean"/>
|
||||
|
||||
<bean class="org.springframework.integration.config.annotation.PublisherAnnotationPostProcessor">
|
||||
<property name="channelRegistry" ref="messageBus"/>
|
||||
</bean>
|
||||
<bean class="org.springframework.integration.config.annotation.PublisherAnnotationPostProcessor"/>
|
||||
|
||||
</beans>
|
||||
|
||||
@@ -19,6 +19,7 @@ package org.springframework.integration.channel;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
@@ -26,13 +27,18 @@ import org.springframework.util.Assert;
|
||||
*/
|
||||
public class TestChannelResolver implements ChannelResolver {
|
||||
|
||||
private final Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
|
||||
private volatile Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
|
||||
|
||||
|
||||
public MessageChannel resolveChannelName(String channelName) {
|
||||
return this.channels.get(channelName);
|
||||
}
|
||||
|
||||
@Autowired
|
||||
public void setChannels(Map<String, MessageChannel> channels) {
|
||||
this.channels = channels;
|
||||
}
|
||||
|
||||
public void addChannel(MessageChannel channel) {
|
||||
Assert.notNull(channel, "'channel' must not be null");
|
||||
Assert.notNull(channel.getName(), "channel name must not be null");
|
||||
|
||||
Reference in New Issue
Block a user