INT-3887: Add receiveTimeout into the @Poller

JIRA: https://jira.spring.io/browse/INT-3887

spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -39,6 +39,7 @@ import org.springframework.scheduling.support.PeriodicTrigger;
  *
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 4.0
  */
 @Target({})
@@ -92,4 +93,15 @@ public @interface Poller {
 	 */
 	String errorChannel() default "";
 
+	/**
+	 * Only applies to polling consumers.
+	 * @return the time the poll thread will wait after the trigger for a new message to arrive.
+	 * Defaults to 1000 (1 second).
+	 * For polled inbound channel adapters, whether or not the polling thread blocks
+	 * is dependent on the message source implementation.
+	 * Can be specified as 'property placeholder', e.g. {@code ${my.poller.receiveTimeout}}.
+	 * @since 5.1
+	 */
+	String receiveTimeout() default "";
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java

@@ -364,6 +364,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			String maxMessagesPerPollValue = this.beanFactory.resolveEmbeddedValue(poller.maxMessagesPerPoll());
 			String cron = this.beanFactory.resolveEmbeddedValue(poller.cron());
 			String errorChannel = this.beanFactory.resolveEmbeddedValue(poller.errorChannel());
+			String receiveTimeout = this.beanFactory.resolveEmbeddedValue(poller.receiveTimeout());
 
 			if (StringUtils.hasText(ref)) {
 				Assert.state(!StringUtils.hasText(triggerRef) && !StringUtils.hasText(executorRef) &&
@@ -416,6 +417,10 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 					errorHandler.setBeanFactory(this.beanFactory);
 					pollerMetadata.setErrorHandler(errorHandler);
 				}
+
+				if (StringUtils.hasText(receiveTimeout)) {
+					pollerMetadata.setReceiveTimeout(Long.parseLong(receiveTimeout));
+				}
 			}
 		}
 		else {
@@ -499,8 +504,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			for (Annotation annotation : annotations) {
 				Object value = AnnotationUtils.getValue(annotation, attribute);
 				if (MessagingAnnotationUtils.hasValue(value)) {
-					throw new BeanDefinitionValidationException("The MessageHandler [" + handlerBeanName + "] can not" +
-							" be populated because of ambiguity with annotation attributes " +
+					throw new BeanDefinitionValidationException("The MessageHandler [" + handlerBeanName +
+							"] can not be populated because of ambiguity with annotation attributes " +
 							this.messageHandlerAttributes + " which are not allowed when an integration annotation " +
 							"is used with a @Bean definition for a MessageHandler." +
 							"\nThe attribute causing the ambiguity is: [" + attribute + "]." +

spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java

@@ -149,6 +149,7 @@ import reactor.core.publisher.Mono;
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 4.0
  */
 @ContextConfiguration(loader = NoBeansOverrideAnnotationConfigContextLoader.class,
@@ -295,7 +296,6 @@ public class EnableIntegrationTests {
 
 	@Test
 	public void testAnnotatedServiceActivator() throws Exception {
-		this.serviceActivatorEndpoint.setReceiveTimeout(10000);
 		this.serviceActivatorEndpoint.start();
 		assertTrue(this.inputReceiveLatch.await(10, TimeUnit.SECONDS));
 
@@ -1239,7 +1239,9 @@ public class EnableIntegrationTests {
 
 		@Override
 		@ServiceActivator(inputChannel = "input", outputChannel = "output", autoStartup = "false",
-				poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", fixedDelay = "${poller.interval}"))
+				poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}",
+						fixedDelay = "${poller.interval}",
+						receiveTimeout = "${poller.receiveTimeout}"))
 		@Publisher
 		@Payload("#args[0].toLowerCase()")
 		@Role("foo")
@@ -1249,7 +1251,8 @@ public class EnableIntegrationTests {
 
 		@Override
 		@ServiceActivator(inputChannel = "input1", outputChannel = "output",
-				poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", fixedRate = "${poller.interval}"))
+				poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}",
+						fixedRate = "${poller.interval}"))
 		@Publisher
 		@Payload("#args[0].toLowerCase()")
 		public String handle1(String payload) {
@@ -1285,7 +1288,8 @@ public class EnableIntegrationTests {
 		/*
 		 * This is an error because input5 is not defined and is therefore a DirectChannel.
 		 */
-		/*@ServiceActivator(inputChannel = "input5", outputChannel = "output", poller = @Poller("defaultPollerMetadata"))
+		/*@ServiceActivator(inputChannel = "input5", outputChannel = "output",
+							poller = @Poller("defaultPollerMetadata"))
 		@Publisher
 		@Payload("#args[0].toLowerCase()")
 		public String handle5(String payload) {
@@ -1669,8 +1673,9 @@ public class EnableIntegrationTests {
 	}
 
 	// Error because the annotation is on a class; it must be on an interface
-//	@MessagingGateway(defaultRequestChannel = "gatewayChannel", defaultHeaders = @GatewayHeader(name = "foo", value = "FOO"))
-//	public static class TestGateway2 { }
+	//	@MessagingGateway(defaultRequestChannel = "gatewayChannel",
+	//        defaultHeaders = @GatewayHeader(name = "foo", value = "FOO"))
+	//	public static class TestGateway2 { }
 
 
 	public static class TestSpelFunction {

spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.properties

@@ -1,4 +1,5 @@
 message.history.tracked.components=input, publishedChannel, annotationTestService*
 poller.maxMessagesPerPoll=10
 poller.interval=100
+poller.receiveTimeout=10000
 global.wireTap.pattern=input