GH-2280 Add per-binding poller configuration

docs/src/main/asciidoc/spring-cloud-stream.adoc

@@ -606,6 +606,15 @@ For example `--spring.cloud.stream.poller.fixed-delay=2000` sets the poller inte
 IMPORTANT: These poller properties are deprecated starting version 3.2 in favor of similar configuration properties from Spring Boot auto-configuration for Spring Integration.
 See `org.springframework.boot.autoconfigure.integration.IntegrationProperties.Poller` for more information.
 
+====== Per-binding polling configuration
+
+The previous section shows how to configure a single default poller that will be applied to all bindings. While it fits well with the model of microservices spring-cloud-stream designed for where each microservice represents a single component (e.g., Supplier) and thus default poller configuration is enough, there are edge cases where
+you may have several components that require different polling configurations 
+
+For such cases please use per-binding way of configuring poller. For example, assume you have an output binding `supply-out-0`. In this case you can configure poller for such
+binding using `spring.cloud.stream.bindings.supply-out-0.producer.poller..` prefix (e.g., `spring.cloud.stream.bindings.supply-out-0.producer.poller.fixed-delay=2000`).
+
+
 ===== Sending arbitrary data to an output (e.g. Foreign event-driven sources)
 
 There are cases where the actual source of data may be coming from the external (foreign) system that is not a binder. For example, the

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.binder;
 
 import java.io.IOException;
+import java.time.Duration;
 
 import javax.validation.constraints.Min;
 
@@ -62,6 +63,8 @@ public class ProducerProperties {
 	 */
 	private String partitionSelectorName;
 
+	private PollerProperties poller;
+
 	@JsonSerialize(using = ExpressionSerializer.class)
 	private Expression partitionSelectorExpression;
 
@@ -169,6 +172,14 @@ public class ProducerProperties {
 		this.autoStartup = autoStartup;
 	}
 
+	public PollerProperties getPoller() {
+		return poller;
+	}
+
+	public void setPoller(PollerProperties poller) {
+		this.poller = poller;
+	}
+
 	static class ExpressionSerializer extends JsonSerializer<Expression> {
 
 		@Override
@@ -181,4 +192,47 @@ public class ProducerProperties {
 
 	}
 
+	public static class PollerProperties {
+
+		private Duration fixedDelay = Duration.ofMillis(1000);
+
+		private long maxMessagesPerPoll = 1L;
+
+		private String cron;
+
+		private Duration initialDelay = Duration.ofMillis(0);
+
+		public long getMaxMessagesPerPoll() {
+			return maxMessagesPerPoll;
+		}
+
+		public void setMaxMessagesPerPoll(long maxMessagesPerPoll) {
+			this.maxMessagesPerPoll = maxMessagesPerPoll;
+		}
+
+		public String getCron() {
+			return cron;
+		}
+
+		public void setCron(String cron) {
+			this.cron = cron;
+		}
+
+		public Duration getInitialDelay() {
+			return initialDelay;
+		}
+
+		public void setInitialDelay(Duration initialDelay) {
+			this.initialDelay = initialDelay;
+		}
+
+		public Duration getFixedDelay() {
+			return fixedDelay;
+		}
+
+		public void setFixedDelay(Duration fixedDelay) {
+			this.fixedDelay = fixedDelay;
+		}
+	}
+
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -21,6 +21,7 @@ import java.lang.reflect.GenericArrayType;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -53,6 +54,7 @@ import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.cloud.function.cloudevent.CloudEventMessageUtils;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.FunctionProperties;
@@ -70,6 +72,7 @@ import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binder.BindingCreatedEvent;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties.PollerProperties;
 import org.springframework.cloud.stream.binding.BindableProxyFactory;
 import org.springframework.cloud.stream.binding.BinderAwareChannelResolver.NewDestinationBindingCallback;
 import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration;
@@ -96,6 +99,7 @@ import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.util.IntegrationReactiveUtils;
 import org.springframework.lang.Nullable;
@@ -105,6 +109,9 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.Trigger;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
@@ -311,11 +318,25 @@ public class FunctionConfiguration {
 			taskScheduler.schedule(() -> { }, Instant.now()); // will keep AC alive
 		}
 		else { // implies pollable
+			AtomicReference<PollerMetadata> pollerMetadata = new AtomicReference<>();
+			if (producerProperties != null && producerProperties.getPoller() != null) {
+				PollerProperties poller = producerProperties.getPoller();
+
+				PollerMetadata pm = new PollerMetadata();
+				PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+				map.from(poller::getMaxMessagesPerPoll).to(pm::setMaxMessagesPerPoll);
+				map.from(poller).as(this::asTrigger).to(pm::setTrigger);
+				pollerMetadata.set(pm);
+			}
 
 			boolean autoStartup = producerProperties != null ? producerProperties.isAutoStartup() : true;
-			integrationFlowBuilder = IntegrationFlows
-					.fromSupplier(supplier, spca -> spca.id(bindingName + "_spca").autoStartup(autoStartup));
-			//only apply the PollableBean attributes if this is a reactive function.
+			integrationFlowBuilder = pollerMetadata == null
+					? IntegrationFlows.fromSupplier(supplier,
+							spca -> spca.id(bindingName + "_spca").autoStartup(autoStartup))
+					: IntegrationFlows.fromSupplier(supplier, spca -> spca.id(bindingName + "_spca")
+							.poller(pollerMetadata.get()).autoStartup(autoStartup));
+
+			// only apply the PollableBean attributes if this is a reactive function.
 			if (splittable && reactive) {
 				integrationFlowBuilder = integrationFlowBuilder.split();
 			}
@@ -324,6 +345,21 @@ public class FunctionConfiguration {
 		return integrationFlowBuilder;
 	}
 
+	private Trigger asTrigger(PollerProperties poller) {
+		if (StringUtils.hasText(poller.getCron())) {
+			return new CronTrigger(poller.getCron());
+		}
+		return createPeriodicTrigger(poller.getFixedDelay(), poller.getInitialDelay());
+	}
+
+	private Trigger createPeriodicTrigger(Duration period, Duration initialDelay) {
+		PeriodicTrigger trigger = new PeriodicTrigger(period.toMillis());
+		if (initialDelay != null) {
+			trigger.setInitialDelay(initialDelay.toMillis());
+		}
+		return trigger;
+	}
+
 	private PollableBean extractPollableAnnotation(StreamFunctionProperties functionProperties, GenericApplicationContext context,
 			BindableFunctionProxyFactory proxyFactory) {
 		// here we need to ensure that for cases where composition is defined we only look for supplier method to find Pollable annotation.

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java

@@ -24,6 +24,7 @@ import org.junit.jupiter.api.Test;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -68,15 +69,26 @@ public class ExplicitBindingTests {
 	}
 
 	@Test
-	public void testExplicitBindingsWithExistingSupplier() {
+	void testExplicitBindingsWithExistingSupplier() {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class))
 						.web(WebApplicationType.NONE)
 						.run("--spring.jmx.enabled=false",
-								"--spring.cloud.stream.input-bindings=supply")) {
+								"--spring.cloud.stream.input-bindings=supply",
+								"--spring.cloud.stream.bindings.supply-out-0.producer.poller.fixed-delay=3000")) {
+
 
 			assertThat(context.getBean("supply-in-0", MessageChannel.class)).isNotNull();
 			assertThat(context.getBean("supply-out-0", MessageChannel.class)).isNotNull();
+
+			OutputDestination output = context.getBean(OutputDestination.class);
+			assertThat(output.receive()).isNotNull();
+			assertThat(output.receive(500)).isNull();
+			assertThat(output.receive(500)).isNull();
+			assertThat(output.receive(500)).isNull();
+			assertThat(output.receive(500)).isNull();
+			assertThat(output.receive(500)).isNull();
+			assertThat(output.receive(500)).isNotNull();
 		}
 	}