From 58a7a74cb49a1594822d964d32c8504fb1905f6d Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 16 Mar 2022 15:50:55 +0100 Subject: [PATCH] GH-2280 Add per-binding poller configuration --- .../main/asciidoc/spring-cloud-stream.adoc | 9 ++++ .../stream/binder/ProducerProperties.java | 54 +++++++++++++++++++ .../function/FunctionConfiguration.java | 42 +++++++++++++-- .../stream/binding/ExplicitBindingTests.java | 16 +++++- 4 files changed, 116 insertions(+), 5 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 70d9b562c..df4d325f7 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java index ec78451c3..df13291a7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java +++ b/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 { @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; + } + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index aa625409d..6fdd20f86 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/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 = 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. diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java index 8d3788844..cfd48fbde 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java +++ b/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(); } }