diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index f8e9a7594..982f93b6a 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -485,6 +485,102 @@ NOTE: We are using `--spring.cloud.function.definition` property to explicitly d we want to be bound to binding destinations. For cases when you only have single such bean it is not required but for all other cases it is. +===== Suppliers (Sources) + +`Function` and `Consumer` are pretty straightforward when it comes to how their invocation is triggered. They are triggered based +on data (events) sent to the destination they are bound to. In other words, they are classic event-driven components. + +However, `Supplier` is in its own category when it comes to triggering. Since it is, by definition, the source (the origin) of the data, it does not +subscribe to any in-bound destination and, therefore, has to be triggered by some other mechanism(s). +There is also a question of `Supplier` implementation, which could be _imperative_ or _reactive_ and which directly relates to the triggering of such suppliers. + +Consider the following sample: + +[source,java] +---- +@SpringBootApplication +public static class SupplierConfiguration { + + @Bean + public Supplier stringSupplier() { + return () -> "Hello from Supplier"; + } +} +---- + +The preceding `Supplier` bean produces a string whenever its `get()` method is invoked. However, who invokes this method and how often? +The framework provides a default polling mechanism (answering the question of "Who?") that will trigger the invocation of the supplier and by default it will do so +every second (answering the question of "How often?"). +In other words, the above configuration produces a single message every second and each message is sent to an `output` destination that is exposed by the binder. +To learn how to customize the polling mechanism, see <> section. + +Consider a different sample: + +[source,java] +---- +@SpringBootApplication +public static class SupplierConfiguration { + + @Bean + public Supplier> stringSupplier() { + return () -> Flux.from(emitter -> { + while (true) { + try { + emitter.onNext("Hello from Supplier"); + Thread.sleep(1000); + } catch (Exception e) { + // ignore + } + } + }); + } +} +---- + +The preceding `Supplier` bean adopts the reactive programming style. Typically, and unlike the imperative supplier, +it should be triggered only once, given that the invocation of its `get()` method produces (supplies) the continuous stream of messages and not an +individual message. + +The framework recognizes the difference in the programming style and guarantees that such a supplier is triggered only once. + +However, imagine the use case where you want to poll some data source and return a finite stream of data representing the result set. +The reactive programming style is a perfect mechanism for such a Supplier. However, given the finite nature of the produced stream, +such Supplier still needs to be invoked periodically. + +Consider the following sample, which emulates such use case by producing a finite stream of data: + +[source,java] +---- +@SpringBootApplication +public static class SupplierConfiguration { + + @PollableSupplier + public Supplier> stringSupplier() { + return () -> Flux.just("hello", "bye"); + } +} +---- + +The bean itself is annotated with `PollableSupplier` annotation (sub-set of `@Bean`), thus signaling to the framework that although the implementation +of such a supplier is reactive, it still needs to be polled. + +====== Polling Configuration Properties + +The following properties are exposed by `org.springframework.cloud.stream.config.DefaultPollerProperties` and are prefixed with +`spring.cloud.stream.poller`: + +fixedDelay:: +Fixed delay for default poller. ++ +Default: 1000L. + +maxMessagesPerPoll:: +Maximum messages for each polling event of the default poller. ++ +Default: 1L. + +For example `--spring.cloud.stream.poller.fixed-delay=2000` sets the poller interval to poll every two seconds. + ===== Content-based routing with functions Routing with functions can be achieved by relying on `RoutingFunction` available in Spring Cloud Function 3.0. All you need to do is enable it via `--spring.cloud.stream.function.routing.enabled=true` application property. Once enabled `RoutingFunction` will be bound to input destination diff --git a/pom.xml b/pom.xml index 42446fa6d..02ce43fde 100644 --- a/pom.xml +++ b/pom.xml @@ -115,7 +115,6 @@ spring-cloud-stream-test-support spring-cloud-stream-test-support-internal spring-cloud-stream-integration-tests - docs diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DefaultPollerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DefaultPollerProperties.java index 43f1523eb..8344921bf 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DefaultPollerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DefaultPollerProperties.java @@ -22,9 +22,9 @@ import org.springframework.scheduling.support.PeriodicTrigger; /** * @author Dave Syer - * + * @author Oleg Zhurakousky */ -@ConfigurationProperties("spring.integration.poller") +@ConfigurationProperties("spring.cloud.stream.poller") public class DefaultPollerProperties { /** diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index 1db1f9c85..6aabfb257 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -39,9 +39,11 @@ import org.springframework.context.annotation.Bean; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.handler.LoggingHandler; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.support.PeriodicTrigger; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.fail; @@ -247,7 +249,8 @@ public class ImplicitFunctionBindingTests { SingleFunctionConfiguration.class)) .web(WebApplicationType.NONE) .run("--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.content-type=text/plain")) { + "--spring.cloud.stream.bindings.input.content-type=text/plain", + "--debug")) { InputDestination inputDestination = context.getBean(InputDestination.class); OutputDestination outputDestination = context @@ -288,6 +291,26 @@ public class ImplicitFunctionBindingTests { } } + @Test + public void testSupplierWithCustomPoller() { + System.clearProperty("spring.cloud.stream.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SupplierWithExplicitPollerConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.poller.fixed-delay=2000")) { + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + PollerMetadata pollerMetadata = context.getBean(PollerMetadata.class); + assertThat(((PeriodicTrigger) pollerMetadata.getTrigger()).getPeriod()).isEqualTo(2000); + + Message outputMessage = outputDestination.receive(6000); + assertThat(outputMessage.getPayload()).isEqualTo("hello".getBytes()); + } + } + @EnableAutoConfiguration public static class NoEnableBindingConfiguration { @@ -393,4 +416,14 @@ public class ImplicitFunctionBindingTests { } + + @EnableAutoConfiguration + public static class SupplierWithExplicitPollerConfiguration { + + @Bean + public Supplier supplier() { + return () -> "hello"; + } + } + }