INT-3073: Add FluxAggregatorMessageHandler

JIRA: https://jira.spring.io/browse/INT-3073

Add `FluxAggregatorMessageHandlerTests`

* Add `FluxAggregatorMessageHandlerTests`
* Fix `FluxAggregatorMessageHandler` for the default `messageForWindowFlux`
to rely on the `publish().autoConnect()` to build a target `Flux` for
window and also copy headers from the first `Message` in the window.
Looks like `switchOnFirst()` doesn't work somehow with windows or I just
don't use it properly, although it isn't clear how to continue chain but
get the whole `Flux` as a single entry for message payload

* Add more tests for `FluxAggregatorMessageHandlerTests`
* Fix `AbstractMessageSplitter` to cast properly a primitive array

* Implement `Lifecycle` in the `FluxAggregatorMessageHandler`

* Add JavaDocs into the `FluxAggregatorMessageHandler`

Javadoc polishing

spring-integration-core/src/test/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandlerTests.java

@@ -0,0 +1,295 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.aggregator;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.GenericMessage;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.2
+ */
+@SuppressWarnings("unchecked")
+class FluxAggregatorMessageHandlerTests {
+
+	@Test
+	void testDefaultAggregation() {
+		QueueChannel resultChannel = new QueueChannel();
+		FluxAggregatorMessageHandler fluxAggregatorMessageHandler = new FluxAggregatorMessageHandler();
+		fluxAggregatorMessageHandler.setOutputChannel(resultChannel);
+		fluxAggregatorMessageHandler.start();
+
+		for (int i = 0; i < 20; i++) {
+			Message<?> messageToAggregate =
+					MessageBuilder.withPayload("" + i)
+							.setCorrelationId(i % 2)
+							.setSequenceSize(10)
+							.build();
+			fluxAggregatorMessageHandler.handleMessage(messageToAggregate);
+		}
+
+		Message<?> result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull()
+				.extracting(Message::getHeaders)
+				.satisfies((headers) ->
+						assertThat((MessageHeaders) headers)
+								.containsEntry(IntegrationMessageHeaderAccessor.CORRELATION_ID, 0));
+
+		Object payload = result.getPayload();
+		assertThat(payload).isInstanceOf(Flux.class);
+
+		Flux<Message<?>> window = (Flux<Message<?>>) payload;
+
+		StepVerifier.create(
+				window.map(Message::getPayload)
+						.cast(String.class))
+				.expectNextSequence(
+						IntStream.iterate(0, i -> i + 2)
+								.limit(10)
+								.mapToObj(Objects::toString)
+								.collect(Collectors.toList()))
+				.verifyComplete();
+
+		result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull()
+				.extracting(Message::getHeaders)
+				.satisfies((headers) ->
+						assertThat((MessageHeaders) headers)
+								.containsEntry(IntegrationMessageHeaderAccessor.CORRELATION_ID, 1));
+
+		payload = result.getPayload();
+		window = (Flux<Message<?>>) payload;
+
+		StepVerifier.create(
+				window.map(Message::getPayload)
+						.cast(String.class))
+				.expectNextSequence(
+						IntStream.iterate(1, i -> i + 2)
+								.limit(10)
+								.mapToObj(Objects::toString)
+								.collect(Collectors.toList()))
+				.verifyComplete();
+
+		fluxAggregatorMessageHandler.stop();
+	}
+
+	@Test
+	void testCustomCombineFunction() {
+		QueueChannel resultChannel = new QueueChannel();
+		FluxAggregatorMessageHandler fluxAggregatorMessageHandler = new FluxAggregatorMessageHandler();
+		fluxAggregatorMessageHandler.setOutputChannel(resultChannel);
+		fluxAggregatorMessageHandler.setWindowSize(10);
+		fluxAggregatorMessageHandler.setCombineFunction(
+				(messageFlux) ->
+						messageFlux
+								.map(Message::getPayload)
+								.collectList()
+								.map(GenericMessage::new));
+		fluxAggregatorMessageHandler.start();
+
+		for (int i = 0; i < 20; i++) {
+			Message<?> messageToAggregate =
+					MessageBuilder.withPayload(i)
+							.setCorrelationId(i % 2)
+							.build();
+			fluxAggregatorMessageHandler.handleMessage(messageToAggregate);
+		}
+
+		Message<?> result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull();
+
+		Object payload = result.getPayload();
+		assertThat(payload)
+				.isInstanceOf(List.class)
+				.asList()
+				.containsExactly(
+						IntStream.iterate(0, i -> i + 2)
+								.limit(10)
+								.boxed()
+								.toArray());
+
+		result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull();
+
+		payload = result.getPayload();
+		assertThat(payload)
+				.isInstanceOf(List.class)
+				.asList()
+				.containsExactly(
+						IntStream.iterate(1, i -> i + 2)
+								.limit(10)
+								.boxed()
+								.toArray());
+
+		fluxAggregatorMessageHandler.stop();
+	}
+
+	@Test
+	void testWindowTimespan() {
+		QueueChannel resultChannel = new QueueChannel();
+		FluxAggregatorMessageHandler fluxAggregatorMessageHandler = new FluxAggregatorMessageHandler();
+		fluxAggregatorMessageHandler.setOutputChannel(resultChannel);
+		fluxAggregatorMessageHandler.setWindowTimespan(Duration.ofMillis(100));
+		fluxAggregatorMessageHandler.start();
+
+		Executors.newSingleThreadExecutor()
+				.submit(() -> {
+					for (int i = 0; i < 10; i++) {
+						Message<?> messageToAggregate =
+								MessageBuilder.withPayload(i)
+										.setCorrelationId("1")
+										.build();
+						fluxAggregatorMessageHandler.handleMessage(messageToAggregate);
+						Thread.sleep(20);
+					}
+					return null;
+				});
+
+		Message<?> result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull();
+
+		Flux<Message<?>> window = (Flux<Message<?>>) result.getPayload();
+
+		List<Integer> messageList =
+				window.map(Message::getPayload)
+						.cast(Integer.class)
+						.collectList()
+						.block(Duration.ofSeconds(10));
+
+		assertThat(messageList)
+				.isNotEmpty()
+				.hasSizeLessThan(10)
+				.contains(0, 1);
+
+		result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull();
+
+		window = (Flux<Message<?>>) result.getPayload();
+
+		messageList =
+				window.map(Message::getPayload)
+						.cast(Integer.class)
+						.collectList()
+						.block(Duration.ofSeconds(10));
+
+		assertThat(messageList)
+				.isNotEmpty()
+				.hasSizeLessThan(10)
+				.doesNotContain(0, 1);
+
+		fluxAggregatorMessageHandler.stop();
+	}
+
+	@Test
+	void testBoundaryTrigger() {
+		QueueChannel resultChannel = new QueueChannel();
+		FluxAggregatorMessageHandler fluxAggregatorMessageHandler = new FluxAggregatorMessageHandler();
+		fluxAggregatorMessageHandler.setOutputChannel(resultChannel);
+		fluxAggregatorMessageHandler.setBoundaryTrigger((message) -> "terminate".equals(message.getPayload()));
+		fluxAggregatorMessageHandler.start();
+
+		for (int i = 0; i < 3; i++) {
+			Message<?> messageToAggregate =
+					MessageBuilder.withPayload("" + i)
+							.setCorrelationId("1")
+							.build();
+			fluxAggregatorMessageHandler.handleMessage(messageToAggregate);
+		}
+
+		fluxAggregatorMessageHandler.handleMessage(
+				MessageBuilder.withPayload("terminate")
+						.setCorrelationId("1")
+						.build());
+
+		fluxAggregatorMessageHandler.handleMessage(
+				MessageBuilder.withPayload("next")
+						.setCorrelationId("1")
+						.build());
+
+		Message<?> result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull();
+
+		Flux<Message<?>> window = (Flux<Message<?>>) result.getPayload();
+
+		StepVerifier.create(
+				window.map(Message::getPayload)
+						.cast(String.class))
+				.expectNext("0", "1", "2")
+				.expectNext("terminate")
+				.verifyComplete();
+
+		fluxAggregatorMessageHandler.stop();
+	}
+
+
+	@Test
+	void testCustomWindow() {
+		QueueChannel resultChannel = new QueueChannel();
+		FluxAggregatorMessageHandler fluxAggregatorMessageHandler = new FluxAggregatorMessageHandler();
+		fluxAggregatorMessageHandler.setOutputChannel(resultChannel);
+		fluxAggregatorMessageHandler.setWindowConfigurer((group) ->
+				group.windowWhile((message) ->
+						message.getPayload() instanceof Integer));
+		fluxAggregatorMessageHandler.start();
+
+		for (int i = 0; i < 3; i++) {
+			Message<?> messageToAggregate =
+					MessageBuilder.withPayload(i)
+							.setCorrelationId("1")
+							.build();
+			fluxAggregatorMessageHandler.handleMessage(messageToAggregate);
+		}
+
+		fluxAggregatorMessageHandler.handleMessage(
+				MessageBuilder.withPayload("terminate")
+						.setCorrelationId("1")
+						.build());
+
+		Message<?> result = resultChannel.receive(10_000);
+		assertThat(result).isNotNull();
+
+		Flux<Message<?>> window = (Flux<Message<?>>) result.getPayload();
+
+		StepVerifier.create(
+				window.map(Message::getPayload)
+						.cast(Integer.class))
+				.expectNext(0, 1, 2)
+				.verifyComplete();
+
+		fluxAggregatorMessageHandler.stop();
+	}
+
+}

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

@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -32,6 +33,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.aggregator.FluxAggregatorMessageHandler;
 import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
@@ -40,6 +42,7 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.MessageChannelSpec;
 import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.dsl.Transformers;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
 import org.springframework.integration.handler.MessageTriggerAction;
 import org.springframework.integration.json.ObjectToJsonTransformer;
 import org.springframework.integration.support.MessageBuilder;
@@ -52,6 +55,8 @@ import org.springframework.test.context.junit4.SpringRunner;
 
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.TextNode;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 /**
  * @author Artem Bilan
@@ -167,6 +172,36 @@ public class CorrelationHandlerTests {
 				.hasSize(0);
 	}
 
+	@Autowired
+	private IntegrationFlowContext integrationFlowContext;
+
+	@Test
+	public void testFluxAggregator() {
+		IntegrationFlow testFlow = (flow) ->
+				flow.split()
+						.channel(MessageChannels.flux())
+						.handle(new FluxAggregatorMessageHandler());
+
+		IntegrationFlowContext.IntegrationFlowRegistration registration =
+				this.integrationFlowContext.registration(testFlow)
+						.register();
+
+		@SuppressWarnings("unchecked")
+		Flux<Message<?>> window =
+				registration.getMessagingTemplate()
+						.convertSendAndReceive(new Integer[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 }, Flux.class);
+
+		assertThat(window).isNotNull();
+
+		StepVerifier.create(
+				window.map(Message::getPayload)
+						.cast(Integer.class))
+				.expectNextSequence(IntStream.range(0, 10).boxed().collect(Collectors.toList()))
+				.verifyComplete();
+
+		registration.destroy();
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {