GH-3189: Properly handle async in ScatterGatherHandler (#10469)

* GH-3189: Properly handle `async` in `ScatterGatherHandler`

Fixes: #3189

Despite supporting `async = true`, the `ScatterGatherHandler`
does blocking in its `handleRequestMessage()`
on the `gatherResultChannel.receive()` call.

* Fix `ScatterGatherHandler` to handle an `async` mode via internal `Mono`
 for the reply object
* Use pattern variable expressions for `ifs` in the `ScatterGatherHandler.doInit()`
for the better readability
* Extract `ScatterGatherHandler.replyFromGatherResult()` method to avoid code duplication
* Document a new (fixed) functionality
* Mention `Mono` reply in the `ScatterGatherHandler` Javadocs

**Auto-cherry-pick to `6.5.x`**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java

@@ -16,6 +16,11 @@
 
 package org.springframework.integration.scattergather;
 
+import java.time.Duration;
+
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
 import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanInitializationException;
@@ -30,6 +35,7 @@
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.management.ManageableLifecycle;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -46,6 +52,9 @@
 /**
  * The {@link MessageHandler} implementation for the
  * <a href="https://www.enterpriseintegrationpatterns.com/BroadcastAggregate.html">Scatter-Gather</a> EIP pattern.
+ * <p>
+ * When {@link #setAsync(boolean)} is {@code true}, the {@link ScatterGatherHandler} produces
+ * a {@link Mono} as a reply based on the gather result.
  *
  * @author Artem Bilan
  * @author Abdul Zaheer
@@ -142,11 +151,11 @@ public Message<?> preSend(Message<?> message, MessageChannel channel) {
 								}
 
 							});
-			if (this.gatherChannel instanceof SubscribableChannel) {
-				this.gatherEndpoint = new EventDrivenConsumer((SubscribableChannel) this.gatherChannel, this.gatherer);
+			if (this.gatherChannel instanceof SubscribableChannel subscribableChannel) {
+				this.gatherEndpoint = new EventDrivenConsumer(subscribableChannel, this.gatherer);
 			}
-			else if (this.gatherChannel instanceof PollableChannel) {
-				this.gatherEndpoint = new PollingConsumer((PollableChannel) this.gatherChannel, this.gatherer);
+			else if (this.gatherChannel instanceof PollableChannel pollableChannel) {
+				this.gatherEndpoint = new PollingConsumer(pollableChannel, this.gatherer);
 				((PollingConsumer) this.gatherEndpoint).setReceiveTimeout(this.gatherTimeout);
 			}
 			else if (this.gatherChannel instanceof ReactiveStreamsSubscribableChannel) {
@@ -187,7 +196,18 @@ private Message<?> enhanceScatterReplyMessage(Message<?> message) {
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		MessageHeaders requestMessageHeaders = requestMessage.getHeaders();
-		PollableChannel gatherResultChannel = new QueueChannel();
+		boolean async = isAsync();
+		MessageChannel gatherResultChannel;
+		Sinks.One<Message<?>> replyMono;
+
+		if (async) {
+			replyMono = Sinks.one();
+			gatherResultChannel = (message, timeout) -> replyMono.tryEmitValue(message).isSuccess();
+		}
+		else {
+			replyMono = null;
+			gatherResultChannel = new QueueChannel();
+		}
 
 		Message<?> scatterMessage =
 				getMessageBuilderFactory()
@@ -200,17 +220,28 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 
 		this.messagingTemplate.send(this.scatterChannel, scatterMessage);
 
-		Message<?> gatherResult = gatherResultChannel.receive(this.gatherTimeout);
-		if (gatherResult != null) {
-			return getMessageBuilderFactory()
-					.fromMessage(gatherResult)
-					.removeHeaders(GATHER_RESULT_CHANNEL, ORIGINAL_ERROR_CHANNEL,
-							MessageHeaders.REPLY_CHANNEL, MessageHeaders.ERROR_CHANNEL);
+		if (replyMono != null) {
+			return replyMono.asMono()
+					.map(this::replyFromGatherResult)
+					.timeout(Duration.ofMillis(this.gatherTimeout), Mono.empty());
+		}
+		else {
+			Message<?> gatherResult = ((PollableChannel) gatherResultChannel).receive(this.gatherTimeout);
+			if (gatherResult != null) {
+				return replyFromGatherResult(gatherResult);
+			}
 		}
 
 		return null;
 	}
 
+	private AbstractIntegrationMessageBuilder<?> replyFromGatherResult(Message<?> gatherResult) {
+		return getMessageBuilderFactory()
+				.fromMessage(gatherResult)
+				.removeHeaders(GATHER_RESULT_CHANNEL, ORIGINAL_ERROR_CHANNEL,
+						MessageHeaders.REPLY_CHANNEL, MessageHeaders.ERROR_CHANNEL);
+	}
+
 	@Override
 	public void start() {
 		if (this.gatherEndpoint != null) {
@@ -236,8 +267,8 @@ private static void checkClass(Class<?> gathererClass, String className, String
 			Assert.isAssignable(clazz, gathererClass,
 					() -> "the '" + type + "' must be an " + className + " " + "instance");
 		}
-		catch (ClassNotFoundException e) {
-			throw new IllegalStateException("The class for '" + className + "' cannot be loaded", e);
+		catch (ClassNotFoundException ex) {
+			throw new IllegalStateException("The class for '" + className + "' cannot be loaded", ex);
 		}
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.dsl.routers;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -463,15 +464,21 @@ public void testRouterAsNonLastComponent() {
 	@Test
 	public void testScatterGather() {
 		QueueChannel replyChannel = new QueueChannel();
-		Message<String> request = MessageBuilder.withPayload("foo")
+		Message<String> request = MessageBuilder.withPayload("test")
 				.setReplyChannel(replyChannel)
 				.build();
 		this.scatterGatherFlowInput.send(request);
 		Message<?> bestQuoteMessage = replyChannel.receive(10000);
-		assertThat(bestQuoteMessage).isNotNull();
-		Object payload = bestQuoteMessage.getPayload();
-		assertThat(payload).isInstanceOf(List.class);
-		assertThat(((List<?>) payload).size()).isGreaterThanOrEqualTo(1);
+		assertThat(bestQuoteMessage)
+				.extracting(Message::getPayload)
+				.asInstanceOf(InstanceOfAssertFactories.LIST)
+				.hasSizeGreaterThanOrEqualTo(1)
+				.first()
+				.asInstanceOf(InstanceOfAssertFactories.type(Message.class))
+				.extracting(Message::getHeaders)
+				.asInstanceOf(InstanceOfAssertFactories.MAP)
+				.extractingByKey("gatherResultChannel")
+				.isNotInstanceOf(PollableChannel.class);
 	}
 
 	@Autowired
@@ -859,9 +866,11 @@ public IntegrationFlow scatterGatherFlow() {
 											group.size() == 3 ||
 													group.getMessages()
 															.stream()
-															.anyMatch(m -> (Double) m.getPayload() > 5)),
+															.anyMatch(m -> (Double) m.getPayload() > 5))
+									.outputProcessor(group -> new ArrayList<>(group.getMessages())),
 							scatterGather -> scatterGather
-									.gatherTimeout(10_000));
+									.gatherTimeout(10_000)
+									.async(true));
 		}
 
 		@Bean

src/reference/antora/modules/ROOT/pages/scatter-gather.adoc

@@ -154,6 +154,10 @@ Mutually exclusive with `scatter-channel` attribute.
 <13> The `<aggregator>` options.
 Required.
 
+NOTE: Starting with version `6.5.3`, when a `ScatterGatherHandler` is configured for the `async = true` option, the request message handling thread is not blocked anymore waiting for a gather result on an internal `((PollableChannel) gatherResultChannel).receive(this.gatherTimeout)` operation.
+Instead, a `reactor.core.publisher.Mono` is returned as a reply object based on a gather result eventually produced from the `gatherResultChannel`.
+Such a `Mono` is handled then according to the xref:reactive-streams.adoc#reactive-reply-payload[Reactive Streams support] in the framework.
+
 [[scatter-gather-error-handling]]
 == Error Handling