Added support for @Polled annotation at class-level alongside the @MessageEndpoint annotation (INT-220).

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;
+
 }

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() {

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;
+		}
+	}
+
 }