From 71fbe29da4c2bdcba7bc12d5edbcdc9233bb94e7 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 14 Jun 2019 11:15:33 -0400 Subject: [PATCH] Add Docs for Flux Aggregator * Some test polishing * Fix not properly wrapped code snippet in the `scripting.adoc` Doc polishing --- .../correlation/CorrelationHandlerTests.java | 7 ++- src/reference/asciidoc/aggregator.adoc | 63 +++++++++++++++++++ src/reference/asciidoc/scripting.adoc | 1 + src/reference/asciidoc/whats-new.adoc | 12 +++- 4 files changed, 77 insertions(+), 6 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index 9d2b20b29c..ecedf33cf8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -177,8 +177,9 @@ public class CorrelationHandlerTests { @Test public void testFluxAggregator() { - IntegrationFlow testFlow = (flow) -> - flow.split() + IntegrationFlow testFlow = + (flow) -> flow + .split() .channel(MessageChannels.flux()) .handle(new FluxAggregatorMessageHandler()); @@ -189,7 +190,7 @@ public class CorrelationHandlerTests { @SuppressWarnings("unchecked") Flux> window = registration.getMessagingTemplate() - .convertSendAndReceive(new Integer[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 }, Flux.class); + .convertSendAndReceive(new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 }, Flux.class); assertThat(window).isNotNull(); diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 1492d69344..41fdf334b4 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -865,3 +865,66 @@ For example, the `JdbcMessageStore` has a `region` property, and the `MongoDbMes For more information about the `MessageStore` interface and its implementations, see <<./message-store.adoc#message-store,Message Store>>. ===== + +[[flux-aggregator]] +==== Flux Aggregator + +In version 5.2, the `FluxAggregatorMessageHandler` component has been introduced. +It is based on the Project Reactor `Flux.groupBy()` and `Flux.window()` operators. +The incoming messages are emitted into the `FluxSink` initiated by the `Flux.create()` in the constructor of this component. +If the `outputChannel` is not provided or it is not an instance of `ReactiveStreamsSubscribableChannel`, the subscription to the main `Flux` is done from the `Lifecycle.start()` implementation. +Otherwise it is postponed to the subscription done by the `ReactiveStreamsSubscribableChannel` implementation. +The messages are grouped by the `Flux.groupBy()` using a `CorrelationStrategy` for the group key. +By default, the `IntegrationMessageHeaderAccessor.CORRELATION_ID` header of the message is consulted. + +By default every closed window is released as a `Flux` in payload of a message to produce. +This message contains all the headers from the first message in the window. +This `Flux` in the output message payload must be subscribed and processed downstream. +Such a logic can be customized (or superseded) by the `setCombineFunction(Function>, Mono>>)` configuration option of the `FluxAggregatorMessageHandler`. +For example, if we would like to have a `List` of payloads in the final message, we can configure a `Flux.collectList()` like this: + +==== +[source,java] +---- +fluxAggregatorMessageHandler.setCombineFunction( + (messageFlux) -> + messageFlux + .map(Message::getPayload) + .collectList() + .map(GenericMessage::new)); +---- +==== + +There are several options in the `FluxAggregatorMessageHandler` to select an appropriate window strategy: + +* `setBoundaryTrigger(Predicate>)` - is propagated to the `Flux.windowUntil()` operator. +See its JavaDocs for more information. +Has a precedence over all other window options. +* `setWindowSize(int)` and `setWindowSizeFunction(Function, Integer>)` - is propagated to the `Flux.window(int)` or `windowTimeout(int, Duration)`. +By default a window size is calculated from the first message in group and its `IntegrationMessageHeaderAccessor.SEQUENCE_SIZE` header. +* `setWindowTimespan(Duration)` - is propagated to the `Flux.window(Duration)` or `windowTimeout(int, Duration)` depending in the window size configuration. +* `setWindowConfigurer(Function>, Flux>>>)` - a function to apply a transformation into the grouped fluxes for any custom window operation not covered by the exposed options. + +Since this component is a `MessageHandler` implementation it can simply be used as a `@Bean` definition together with a `@ServiceActivator` messaging annotation. +With Java DSL it can be used from the `.handle()` EIP-method. +The sample below demonstrates how we can register an `IntegrationFlow` at runtime and how a `FluxAggregatorMessageHandler` can be correlated with a splitter upstream: + +==== +[source,java] +---- +IntegrationFlow fluxFlow = + (flow) -> flow + .split() + .channel(MessageChannels.flux()) + .handle(new FluxAggregatorMessageHandler()); + +IntegrationFlowContext.IntegrationFlowRegistration registration = + this.integrationFlowContext.registration(fluxFlow) + .register(); + +@SuppressWarnings("unchecked") +Flux> window = + registration.getMessagingTemplate() + .convertSendAndReceive(new Integer[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 }, Flux.class); +---- +==== diff --git a/src/reference/asciidoc/scripting.adoc b/src/reference/asciidoc/scripting.adoc index fc1d0814dd..92629a03a2 100644 --- a/src/reference/asciidoc/scripting.adoc +++ b/src/reference/asciidoc/scripting.adoc @@ -30,6 +30,7 @@ In addition you need to add a script engine implementation, e.g. JRuby, Jython. Starting with version 5.2, Spring Integration provides a Kotlin Jsr223 support. You need to add these dependencies into your project to make it working: +==== [source, groovy] ---- runtime 'org.jetbrains.kotlin:kotlin-script-util' diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 984556d2c5..b4fedad4dc 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -20,24 +20,30 @@ If you are interested in more details, see the Issue Tracker tickets that were r [[x5.2-new-components]] === New Components -[[x5.2-rateLimitAdvice]] +[[x5.2-rate-limit-advice]] === Rate Limit Advice Support The `RateLimiterRequestHandlerAdvice` is now available for limiting requests rate on handlers. See <<./handler-advice.adoc#rate-limiter-advice,Rate Limiter Advice>> for more information. -[[x5.2-cacheAdvice]] +[[x5.2-cache-advice]] === Caching Advice Support The `CacheRequestHandlerAdvice` is now available for caching request results on handlers. See <<./handler-advice.adoc#cache-advice,Caching Advice>> for more information. -[[x5.2-kotlinScripts]] +[[x5.2-kotlin-scripts]] === Kotlin Scripts Support The JSR223 scripting module now includes a support for Kotlin scripts. See <<./scripting.adoc#scripting,Scripting Support>> for more information. +[[x5.2-flux-aggregator]] +=== Flux Aggregator Support + +The `FluxAggregatorMessageHandler` is now available for grouping and windowing messages logic based on the Project Reactor `Flux` operators. +See <<./aggregator.adoc#flux-aggregator,Flux Aggregator>> for more information. + [[x5.2-general]] === General Changes