Fix reply producing to not block reactive thread

When `DirectChannel` is used for reply producing, the data is
handled on the same thread which has produced it (normally), so
if we have a request-reply afterwards (e.g. `gateway()`), this thread
is blocked waiting for reply.
When the thread is assumed to be non-blocked (e.g. Netty event loop),
the request-reply withing such a thread for the same non-blocking client
causes a deadlock: the thread waits for reply, but at the same time it
supposes to fulfil a synchronization barrier with that reply

* Fix `AbstractMessageProducingHandler.asyncNonReactiveReply()` to use
a `publishOn(Schedulers.boundedElastic())` for reply `Mono` to free
producing thread from potential downstream blocking
* Demonstrate deadlock with a new test in the `RSocketDslTests`;
the original report was against WebFlux, but conditions are really
the same: `reactor-netty` is used as a low-level client

**Cherry-pick to `5.4.x`**

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -55,6 +55,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 /**
  * The base {@link AbstractMessageHandler} implementation for the {@link MessageProducer}.
@@ -362,7 +363,9 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 			else {
 				reactiveReply = Mono.from((Publisher<?>) reply);
 			}
-			reactiveReply.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
+			reactiveReply
+					.publishOn(Schedulers.boundedElastic())
+					.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
 			future = settableListenableFuture;
 		}
 		future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel));

spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-2021 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.rsocket.dsl;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import java.time.Duration;
 import java.util.function.Function;
 
@@ -28,6 +30,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
 import org.springframework.integration.rsocket.ClientRSocketConnector;
 import org.springframework.integration.rsocket.RSocketInteractionModel;
 import org.springframework.integration.rsocket.ServerRSocketConnector;
@@ -74,6 +77,39 @@ public class RSocketDslTests {
 				.verifyComplete();
 	}
 
+	@Autowired
+	IntegrationFlowContext integrationFlowContext;
+
+	@Autowired
+	ClientRSocketConnector clientRSocketConnector;
+
+	@Test
+	void testNoBlockingForReactiveThreads() {
+		IntegrationFlow flow =
+				f -> f
+						.handle(RSockets.outboundGateway("/lowercase")
+								.clientRSocketConnector(this.clientRSocketConnector))
+						.transform("{ firstResult: payload }")
+						.enrich(e -> e
+								.requestPayloadExpression("payload.firstResult")
+								.requestSubFlow(
+										sf -> sf
+												.handle(RSockets.outboundGateway("/lowercase")
+														.clientRSocketConnector(this.clientRSocketConnector)))
+								.propertyExpression("secondResult", "payload"))
+						.transform("payload.values().toString()");
+
+		IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
+				this.integrationFlowContext.registration(flow).register();
+
+		String result = flowRegistration.getMessagingTemplate().convertSendAndReceive("TEST", String.class);
+
+		assertThat(result).isEqualTo("[test, test]");
+
+		flowRegistration.destroy();
+	}
+
+
 	@Configuration
 	@EnableIntegration
 	public static class TestConfiguration {
@@ -96,7 +132,7 @@ public class RSocketDslTests {
 			return IntegrationFlows
 					.from(Function.class)
 					.handle(RSockets.outboundGateway(message ->
-							message.getHeaders().getOrDefault("route", "/uppercase"))
+											message.getHeaders().getOrDefault("route", "/uppercase"))
 									.interactionModel((message) -> RSocketInteractionModel.requestChannel)
 									.expectedResponseType("T(java.lang.String)")
 									.clientRSocketConnector(clientRSocketConnector),
@@ -126,6 +162,14 @@ public class RSocketDslTests {
 					.get();
 		}
 
+		@Bean
+		public IntegrationFlow rsocketLowerCaseFlow() {
+			return IntegrationFlows
+					.from(RSockets.inboundGateway("/lowercase"))
+					.<Flux<String>, Flux<String>>transform((flux) -> flux.map(String::toLowerCase))
+					.get();
+		}
+
 	}
 
 }