From 5ab34d7da3beab9738b41011f2f406ee43e13f97 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 20 May 2008 14:08:36 +0000 Subject: [PATCH] Added support for @Polled annotation at class-level alongside the @MessageEndpoint annotation (INT-220). --- .../integration/annotation/Polled.java | 16 ++++++--- ...essageEndpointAnnotationPostProcessor.java | 14 ++++++-- ...eEndpointAnnotationPostProcessorTests.java | 33 +++++++++++++++++++ 3 files changed, 55 insertions(+), 8 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java index c1ebbf2e2a..49deba602a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java @@ -22,27 +22,33 @@ import java.lang.annotation.Inherited; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; +import java.util.concurrent.TimeUnit; import org.springframework.integration.scheduling.PollingSchedule; /** - * Indicates that a method is capable of providing messages. The method must not - * accept any parameters but can return either a single object or collection. - * The enclosing class should be annotated with + * Annotation that can be specified at class-level alongside a + * {@link MessageEndpoint @MessageEndpoint} annotation in order to provide the + * scheduling information for that endpoint. Alternatively, as a method-level + * annotation, this indicates that a method is capable of providing messages. + * The method must not accept any parameters but can return either a single + * object or collection. The enclosing class should be annotated with * {@link MessageEndpoint @MessageEndpoint}. * * @author Mark Fisher */ -@Target(ElementType.METHOD) +@Target({ElementType.TYPE, ElementType.METHOD}) @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented public @interface Polled { - int period() default 1000; + int period() default 0; long initialDelay() default PollingSchedule.DEFAULT_INITIAL_DELAY; boolean fixedRate() default PollingSchedule.DEFAULT_FIXED_RATE; + TimeUnit timeUnit() default TimeUnit.MILLISECONDS; + } 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 fc1f26114e..7cc24ede1c 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 @@ -121,7 +121,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor throw new ConfigurationException("@MessageEndpoint has no handler method"); } HandlerEndpoint endpoint = new HandlerEndpoint(handlerChain); - this.configureInput(bean, beanName, endpointAnnotation, endpoint); + Polled polledAnnotation = AnnotationUtils.findAnnotation(beanClass, Polled.class); + this.configureInput(bean, beanName, endpointAnnotation, polledAnnotation, endpoint); if (StringUtils.hasText(outputChannelName)) { endpoint.setOutputChannelName(outputChannelName); } @@ -142,10 +143,17 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor } private void configureInput(final Object bean, final String beanName, MessageEndpoint annotation, - final HandlerEndpoint endpoint) { + Polled polledAnnotation, final HandlerEndpoint endpoint) { String channelName = annotation.input(); if (StringUtils.hasText(channelName)) { - Subscription subscription = new Subscription(channelName); + PollingSchedule schedule = null; + if (polledAnnotation != null) { + schedule = new PollingSchedule(polledAnnotation.period()); + schedule.setInitialDelay(polledAnnotation.initialDelay()); + schedule.setFixedRate(polledAnnotation.fixedRate()); + schedule.setTimeUnit(polledAnnotation.timeUnit()); + } + Subscription subscription = new Subscription(channelName, schedule); endpoint.setSubscription(subscription); } ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { 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 890594aaff..f0ec3594ae 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 @@ -45,6 +45,8 @@ import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; /** * @author Mark Fisher @@ -316,6 +318,26 @@ public class MessageEndpointAnnotationPostProcessorTests { postProcessor.postProcessAfterInitialization(endpoint, "endpoint"); } + @Test + public void testEndpointWithPolledAnnotation() { + MessageBus messageBus = new MessageBus(); + QueueChannel testChannel = new QueueChannel(); + messageBus.registerChannel("testChannel", testChannel); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + AnnotatedEndpointWithPolledAnnotation endpoint = new AnnotatedEndpointWithPolledAnnotation(); + postProcessor.postProcessAfterInitialization(endpoint, "testBean"); + HandlerEndpoint processedEndpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean-endpoint"); + Schedule schedule = processedEndpoint.getSubscription().getSchedule(); + assertEquals(PollingSchedule.class, schedule.getClass()); + PollingSchedule pollingSchedule = (PollingSchedule) schedule; + assertEquals(1234, pollingSchedule.getPeriod()); + assertEquals(5678, pollingSchedule.getInitialDelay()); + assertEquals(true, pollingSchedule.getFixedRate()); + assertEquals(TimeUnit.SECONDS, pollingSchedule.getTimeUnit()); + } + @MessageEndpoint(output="testChannel") private static class PolledAnnotationTestBean { @@ -425,4 +447,15 @@ public class MessageEndpointAnnotationPostProcessorTests { private static class AnnotatedEndpointWithNoHandlerMethod { } + + @MessageEndpoint(input="testChannel") + @Polled(period=1234, initialDelay=5678, fixedRate=true, timeUnit=TimeUnit.SECONDS) + private static class AnnotatedEndpointWithPolledAnnotation { + + @Handler + public String prependFoo(String s) { + return "foo" + s; + } + } + }