From 07edd8f4239fab30fc7a22cb0bbde228ab1ec325 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 3 Apr 2008 20:31:24 +0000 Subject: [PATCH] MessageEndpointAnnotationPostProcessor now handles proxies and also detects inherited @MessageEndpoint annotations (INT-160). --- ...essageEndpointAnnotationPostProcessor.java | 18 ++- .../PublisherAnnotationPostProcessor.java | 5 +- .../SubscriberAnnotationPostProcessor.java | 7 +- .../endpoint/annotation/ITestEndpoint.java | 26 ++++ ...eEndpointAnnotationPostProcessorTests.java | 127 +++++++++++++++++- .../annotation/SimpleAnnotatedEndpoint.java | 4 +- 6 files changed, 171 insertions(+), 16 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/ITestEndpoint.java 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 a8b679732d..78d8d87c46 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap; 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.InitializingBean; import org.springframework.beans.factory.config.BeanPostProcessor; @@ -100,7 +101,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor } public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - MessageEndpoint endpointAnnotation = bean.getClass().getAnnotation(MessageEndpoint.class); + Class beanClass = this.getBeanClass(bean); + MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class); if (endpointAnnotation == null) { return bean; } @@ -111,7 +113,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handlerChain); this.configureInput(bean, beanName, endpointAnnotation, endpoint); this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint); - Concurrency concurrencyAnnotation = bean.getClass().getAnnotation(Concurrency.class); + Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(beanClass, Concurrency.class); if (concurrencyAnnotation != null) { ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy( concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize()); @@ -138,7 +140,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor Subscription subscription = new Subscription(channelName, schedule); endpoint.setSubscription(subscription); } - ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { + ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class); if (annotation != null) { @@ -173,7 +175,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor endpoint.setDefaultOutputChannelName(channelName); return; } - ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { + ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { boolean foundDefaultOutput = false; public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class); @@ -208,7 +210,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor @SuppressWarnings("unchecked") private MessageHandlerChain createHandlerChain(final Object bean) { final List handlers = new ArrayList(); - ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { + ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Annotation[] annotations = AnnotationUtils.getAnnotations(method); for (Annotation annotation : annotations) { @@ -253,6 +255,10 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor return null; } + private Class getBeanClass(Object bean) { + return AopUtils.getTargetClass(bean); + } + private boolean isHandlerAnnotation(Annotation annotation) { return annotation.annotationType().equals(Handler.class) || annotation.annotationType().isAnnotationPresent(Handler.class) 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 c0347879e9..e676befc06 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -82,8 +82,7 @@ public class PublisherAnnotationPostProcessor implements BeanPostProcessor, Bean } public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - Class targetClass = bean instanceof Advised ? - ((Advised) bean).getTargetSource().getTargetClass() : bean.getClass(); + Class targetClass = AopUtils.getTargetClass(bean); if (targetClass == null) { return bean; } 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 33c2b2b423..6338a14627 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -22,7 +22,7 @@ import java.lang.reflect.Method; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.aop.framework.Advised; +import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.core.annotation.AnnotationUtils; @@ -71,8 +71,7 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { } public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException { - final Class targetClass = bean instanceof Advised ? - ((Advised) bean).getTargetSource().getTargetClass() : bean.getClass(); + final Class targetClass = AopUtils.getTargetClass(bean); if (targetClass == null) { return bean; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/ITestEndpoint.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/ITestEndpoint.java new file mode 100644 index 0000000000..8434e4690c --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/ITestEndpoint.java @@ -0,0 +1,26 @@ +/* + * 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.endpoint.annotation; + +/** + * @author Mark Fisher + */ +public interface ITestEndpoint { + + String sayHello(String name); + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java index a91a93640a..fd5c920d0b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -25,10 +25,12 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.springframework.aop.framework.ProxyFactory; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.annotation.Concurrency; import org.springframework.integration.annotation.DefaultOutput; +import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Polled; import org.springframework.integration.bus.MessageBus; @@ -149,6 +151,110 @@ public class MessageEndpointAnnotationPostProcessorTests { assertEquals(messageBus, channelRegistry); } + @Test + public void testProxiedMessageEndpointAnnotation() { + MessageBus messageBus = new MessageBus(); + messageBus.setAutoCreateChannels(true); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpoint()); + Object proxy = proxyFactory.getProxy(); + postProcessor.postProcessAfterInitialization(proxy, "proxy"); + messageBus.start(); + MessageChannel inputChannel = messageBus.lookupChannel("inputChannel"); + MessageChannel outputChannel = messageBus.lookupChannel("outputChannel"); + inputChannel.send(new StringMessage("world")); + Message message = outputChannel.receive(1000); + assertEquals("hello world", message.getPayload()); + } + + @Test + public void testMessageEndpointAnnotationInherited() { + MessageBus messageBus = new MessageBus(); + messageBus.setAutoCreateChannels(true); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointSubclass(), "subclass"); + messageBus.start(); + MessageChannel inputChannel = messageBus.lookupChannel("inputChannel"); + MessageChannel outputChannel = messageBus.lookupChannel("outputChannel"); + inputChannel.send(new StringMessage("world")); + Message message = outputChannel.receive(1000); + assertEquals("hello world", message.getPayload()); + } + + @Test + public void testMessageEndpointAnnotationInheritedWithProxy() { + MessageBus messageBus = new MessageBus(); + messageBus.setAutoCreateChannels(true); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointSubclass()); + Object proxy = proxyFactory.getProxy(); + postProcessor.postProcessAfterInitialization(proxy, "proxy"); + messageBus.start(); + MessageChannel inputChannel = messageBus.lookupChannel("inputChannel"); + MessageChannel outputChannel = messageBus.lookupChannel("outputChannel"); + inputChannel.send(new StringMessage("world")); + Message message = outputChannel.receive(1000); + assertEquals("hello world", message.getPayload()); + } + + @Test + public void testMessageEndpointAnnotationInheritedFromInterface() { + MessageBus messageBus = new MessageBus(); + MessageChannel inputChannel = new SimpleChannel(); + MessageChannel outputChannel = new SimpleChannel(); + messageBus.registerChannel("inputChannel", inputChannel); + messageBus.registerChannel("outputChannel", outputChannel); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl"); + messageBus.start(); + inputChannel.send(new StringMessage("ABC")); + Message message = outputChannel.receive(1000); + assertEquals("test-ABC", message.getPayload()); + } + + @Test + public void testMessageEndpointAnnotationInheritedFromInterfaceWithAutoCreatedChannels() { + MessageBus messageBus = new MessageBus(); + messageBus.setAutoCreateChannels(true); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl"); + messageBus.start(); + MessageChannel inputChannel = messageBus.lookupChannel("inputChannel"); + MessageChannel outputChannel = messageBus.lookupChannel("outputChannel"); + inputChannel.send(new StringMessage("ABC")); + Message message = outputChannel.receive(1000); + assertEquals("test-ABC", message.getPayload()); + } + + @Test + public void testMessageEndpointAnnotationInheritedFromInterfaceWithProxy() { + MessageBus messageBus = new MessageBus(); + MessageChannel inputChannel = new SimpleChannel(); + MessageChannel outputChannel = new SimpleChannel(); + messageBus.registerChannel("inputChannel", inputChannel); + messageBus.registerChannel("outputChannel", outputChannel); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointImplementation()); + Object proxy = proxyFactory.getProxy(); + postProcessor.postProcessAfterInitialization(proxy, "proxy"); + messageBus.start(); + inputChannel.send(new StringMessage("ABC")); + Message message = outputChannel.receive(1000); + assertEquals("test-ABC", message.getPayload()); + } + @MessageEndpoint(defaultOutput="testChannel") private static class PolledAnnotationTestBean { @@ -204,4 +310,23 @@ public class MessageEndpointAnnotationPostProcessorTests { } } + + private static class SimpleAnnotatedEndpointSubclass extends SimpleAnnotatedEndpoint { + } + + + @MessageEndpoint(input="inputChannel", defaultOutput="outputChannel", pollPeriod=25) + private static interface SimpleAnnotatedEndpointInterface { + String test(String input); + } + + + private static class SimpleAnnotatedEndpointImplementation implements SimpleAnnotatedEndpointInterface { + + @Handler + public String test(String input) { + return "test-" + input; + } + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java index 67691f85a3..3e7a5449e6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -23,7 +23,7 @@ import org.springframework.integration.annotation.MessageEndpoint; * @author Mark Fisher */ @MessageEndpoint(input="inputChannel", defaultOutput="outputChannel", pollPeriod=10) -public class SimpleAnnotatedEndpoint { +public class SimpleAnnotatedEndpoint implements ITestEndpoint { @Handler public String sayHello(String name) {