Add Docs for Flux Aggregator

* Some test polishing
* Fix not properly wrapped code snippet in the `scripting.adoc`

Doc polishing

Author: artembilan

spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java

@@ -177,8 +177,9 @@ public void testSplitterDiscard() {
 
 	@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 void testFluxAggregator() {
 		@SuppressWarnings("unchecked")
 		Flux<Message<?>> 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();
 

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<Flux<Message<?>>, Mono<Message<?>>>)` 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<Message<?>>)` - 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<Message<?>, 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<Message<?>>, Flux<Flux<Message<?>>>>)` - 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<Message<?>> window =
+        registration.getMessagingTemplate()
+                .convertSendAndReceive(new Integer[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 }, Flux.class);
+----
+====

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'

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