From ead292fd1f08fa2b0169cd6c07c606efb4ac2477 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 12 Aug 2008 20:49:55 +0000 Subject: [PATCH] The @MessageTarget annotation is now used with @ChannelAdapter (like @PollableSource) instead of @MessageEndpoint. --- .../TargetAnnotationPostProcessor.java | 37 ++++++------------- ...MessagingAnnotationPostProcessorTests.java | 5 +-- 2 files changed, 13 insertions(+), 29 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java index 9efa52c312..dc2535dc61 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java @@ -22,16 +22,12 @@ import java.util.List; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.ConfigurationException; -import org.springframework.integration.annotation.Concurrency; +import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.PollableChannelAdapter; import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.TargetEndpoint; -import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; import org.springframework.integration.handler.MethodInvokingTarget; import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.scheduling.Schedule; -import org.springframework.util.StringUtils; /** * Post-processor for classes annotated with {@link MessageTarget @MessageTarget}. @@ -45,10 +41,16 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP } - public MessageTarget processMethod(Object bean, Method method, Annotation annotation) { + protected MessageTarget processMethod(Object bean, Method method, Annotation annotation) { MethodInvokingTarget target = new MethodInvokingTarget(); target.setObject(bean); - target.setMethod(method); + target.setMethodName(method.getName()); + ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class); + if (channelAdapterAnnotation != null) { + String channelName = channelAdapterAnnotation.value(); + PollableChannelAdapter adapter = new PollableChannelAdapter(channelName, null, target); + this.getMessageBus().registerChannel(channelName, adapter); + } return target; } @@ -61,24 +63,7 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP public AbstractEndpoint createEndpoint(Object bean, String beanName, Class originalBeanClass, org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) { - TargetEndpoint endpoint = new TargetEndpoint((MessageTarget) bean); - String inputChannelName = endpointAnnotation.input(); - if (StringUtils.hasText(inputChannelName)) { - endpoint.setInputChannelName(inputChannelName); - } - Schedule schedule = this.extractSchedule(originalBeanClass); - if (schedule != null) { - endpoint.setSchedule(schedule); - } - Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class); - if (concurrencyAnnotation != null) { - ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(), - concurrencyAnnotation.maxSize()); - concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); - concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); - endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy, beanName)); - } - return endpoint; + return null; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index 711dc252c5..2798895078 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -162,14 +162,13 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testTargetAnnotation() throws InterruptedException { MessageBus messageBus = new DefaultMessageBus(); - QueueChannel testChannel = new QueueChannel(); - messageBus.registerChannel("testChannel", testChannel); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); CountDownLatch latch = new CountDownLatch(1); TargetAnnotationTestBean testBean = new TargetAnnotationTestBean(latch); postProcessor.postProcessAfterInitialization(testBean, "testBean"); messageBus.start(); + MessageChannel testChannel = messageBus.lookupChannel("testChannel"); testChannel.send(new StringMessage("foo")); latch.await(1000, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); @@ -399,7 +398,7 @@ public class MessagingAnnotationPostProcessorTests { } - @MessageEndpoint(input="testChannel") + @ChannelAdapter("testChannel") private static class TargetAnnotationTestBean { private String messageText;