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

Removed the @Subscriber annotation now that inputChannel can be configured at Method-level for all other Message-handling annotations.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-03 18:36:03 +00:00
parent 2a8431f37a
commit e4c3ead89f
4 changed files with 7 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Subscriber.java
View File

@@ -1,40 +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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Indicates that a method-invoking handler adapter should delegate to this
* method.
*
* @author Mark Fisher
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Subscriber {
String channel();
}

13
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java
View File

@@ -37,16 +37,12 @@ public class AnnotationDrivenParser implements BeanDefinitionParser {
private static final String PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.PublisherAnnotationPostProcessor";
private static final String SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.SubscriberAnnotationPostProcessor";
private static final String MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.MessagingAnnotationPostProcessor";
public BeanDefinition parse(Element element, ParserContext parserContext) {
this.registerPublisherPostProcessor(parserContext);
this.registerSubscriberPostProcessor(parserContext);
this.registerMessagingAnnotationPostProcessor(parserContext);
return null;
}
@@ -60,15 +56,6 @@ public class AnnotationDrivenParser implements BeanDefinitionParser {
parserContext.registerBeanComponent(bcd);
}
private void registerSubscriberPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
bd.getPropertyValues().addPropertyValue("messageBus",
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
BeanComponentDefinition bcd = new BeanComponentDefinition(
bd, SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
parserContext.registerBeanComponent(bcd);
}
private void registerMessagingAnnotationPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(MessagingAnnotationPostProcessor.class);
bd.getConstructorArgumentValues().addGenericArgumentValue(

113
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java
View File

@@ -1,113 +0,0 @@
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Subscriber;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.DefaultServiceInvoker;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* A {@link BeanPostProcessor} that creates a method-invoking handler adapter
* when it discovers methods annotated with {@link Subscriber @Subscriber}.
*
* @author Mark Fisher
*/
public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
private Log logger = LogFactory.getLog(this.getClass());
private Class<? extends Annotation> subscriberAnnotationType = Subscriber.class;
private String channelNameAttribute = "channel";
private MessageBus messageBus;
public void setSubscriberAnnotationType(Class<? extends Annotation> subscriberAnnotationType) {
Assert.notNull(subscriberAnnotationType, "'subscriberAnnotationType' must not be null");
this.subscriberAnnotationType = subscriberAnnotationType;
}
public void setChannelNameAttribute(String channelNameAttribute) {
Assert.notNull(channelNameAttribute, "'channelNameAttribute' must not be null");
this.channelNameAttribute = channelNameAttribute;
}
public void setMessageBus(MessageBus messageBus) {
Assert.notNull(messageBus, "'messageBus' must not be null");
this.messageBus = messageBus;
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException {
final Class<?> targetClass = AopUtils.getTargetClass(bean);
if (targetClass == null) {
return bean;
}
if (this.messageBus == null) {
if (logger.isWarnEnabled()) {
logger.warn(this.getClass().getSimpleName() + " is disabled since no 'messageBus' was provided");
}
return bean;
}
ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = method.getAnnotation(subscriberAnnotationType);
if (annotation != null) {
String channelName = (String) AnnotationUtils.getValue(annotation, channelNameAttribute);
if (!StringUtils.hasText(channelName)) {
throw new ConfigurationException("no channel name provided for subscriber");
}
DefaultServiceInvoker invoker = new DefaultServiceInvoker(bean, method);
invoker.afterPropertiesSet();
String endpointName = ClassUtils.getShortNameAsProperty(targetClass) +
"." + method.getName() + ".endpoint";
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(invoker);
endpoint.setBeanName(endpointName);
MessageChannel inputChannel = messageBus.lookupChannel(channelName);
if (inputChannel == null) {
throw new ConfigurationException("unable to resolve channel '" + channelName + "' for subscriber");
}
endpoint.setSource(inputChannel);
messageBus.registerEndpoint(endpoint);
}
}
});
return bean;
}
}

65
org.springframework.integration/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java
View File

@@ -19,10 +19,6 @@ package org.springframework.integration.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -31,11 +27,12 @@ import org.junit.Test;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.annotation.Subscriber;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.bus.DefaultMessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.annotation.SubscriberAnnotationPostProcessor;
import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor;
import org.springframework.integration.message.StringMessage;
/**
@@ -53,8 +50,8 @@ public class SubscriberAnnotationPostProcessorTests {
context.registerBeanDefinition("testBean", subscriberDef);
String busBeanName = MessageBusParser.MESSAGE_BUS_BEAN_NAME;
context.registerBeanDefinition(busBeanName, new RootBeanDefinition(DefaultMessageBus.class));
RootBeanDefinition postProcessorDef = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
postProcessorDef.getPropertyValues().addPropertyValue("messageBus", new RuntimeBeanReference(busBeanName));
RootBeanDefinition postProcessorDef = new RootBeanDefinition(MessagingAnnotationPostProcessor.class);
postProcessorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(busBeanName));
context.registerBeanDefinition("postProcessor", postProcessorDef);
context.refresh();
context.start();
@@ -69,34 +66,6 @@ public class SubscriberAnnotationPostProcessorTests {
context.stop();
}
@Test
public void testCustomAnnotation() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
GenericApplicationContext context = new GenericApplicationContext();
context.registerBeanDefinition("testChannel", new RootBeanDefinition(QueueChannel.class));
RootBeanDefinition subscriberDef = new RootBeanDefinition(CustomAnnotationTestBean.class);
subscriberDef.getConstructorArgumentValues().addGenericArgumentValue(latch);
context.registerBeanDefinition("testBean", subscriberDef);
String busBeanName = MessageBusParser.MESSAGE_BUS_BEAN_NAME;
context.registerBeanDefinition(busBeanName, new RootBeanDefinition(DefaultMessageBus.class));
RootBeanDefinition postProcessorDef = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
postProcessorDef.getPropertyValues().addPropertyValue("messageBus", new RuntimeBeanReference(busBeanName));
postProcessorDef.getPropertyValues().addPropertyValue("subscriberAnnotationType", CustomSubscriberAnnotation.class);
postProcessorDef.getPropertyValues().addPropertyValue("channelNameAttribute", "subscribeTo");
context.registerBeanDefinition("postProcessor", postProcessorDef);
context.refresh();
context.start();
CustomAnnotationTestBean testBean = (CustomAnnotationTestBean) context.getBean("testBean");
assertEquals(1, latch.getCount());
assertNull(testBean.getMessageText());
MessageChannel testChannel = (MessageChannel) context.getBean("testChannel");
testChannel.send(new StringMessage("test-456"));
latch.await(1000, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
assertEquals("test-456", testBean.getMessageText());
context.stop();
}
public static class AbstractSubscriberAnnotationTestBean {
@@ -118,38 +87,18 @@ public class SubscriberAnnotationPostProcessorTests {
}
@MessageEndpoint
public static class SubscriberAnnotationTestBean extends SubscriberAnnotationPostProcessorTests.AbstractSubscriberAnnotationTestBean {
public SubscriberAnnotationTestBean(CountDownLatch latch) {
super(latch);
}
@Subscriber(channel="testChannel")
@Handler(inputChannel="testChannel")
public void testMethod(String messageText) {
this.messageText = messageText;
this.countDown();
}
}
public static class CustomAnnotationTestBean extends SubscriberAnnotationPostProcessorTests.AbstractSubscriberAnnotationTestBean {
public CustomAnnotationTestBean(CountDownLatch latch) {
super(latch);
}
@CustomSubscriberAnnotation(subscribeTo="testChannel")
public void testMethod(String messageText) {
this.messageText = messageText;
this.countDown();
}
}
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public static @interface CustomSubscriberAnnotation {
String subscribeTo();
}
}