GH-9538: Rely on the customizeMonoReply() for thread switching

Fixes: #9538

Some applications might not be satisfied with `.publishOn(Schedulers.boundedElastic())`
used by default for `Mono` replies.

* Remove that `.publishOn(Schedulers.boundedElastic())` from the `AbstractMessageProducingHandler`.
Instead, the target project is free to make a choice via `customizeMonoReply()`, e.g.:
```
.handle(RSockets.outboundGateway("/lowercase")
										.clientRSocketConnector(this.clientRSocketConnector),
								endpoint -> endpoint.customizeMonoReply((message, mono) ->
										mono.publishOn(Schedulers.boundedElastic())))
```

Author: artembilan

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

@@ -33,7 +33,6 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.BeanFactory;
@@ -386,7 +385,6 @@ private static CompletableFuture<?> toFutureReply(Object reply, @Nullable Reacti
 			CompletableFuture<Object> replyFuture = new CompletableFuture<>();
 
 			reactiveReply
-					.publishOn(Schedulers.boundedElastic())
 					/*
 					The MonoToCompletableFuture in Project Reactor does not support context propagation,
 					 and it does not suppose to, since there is no guarantee how this Future is going to

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

@@ -21,6 +21,7 @@
 
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -42,7 +43,6 @@
 
 /**
  * @author Artem Bilan
- *
  * @since 5.2
  */
 @SpringJUnitConfig
@@ -86,7 +86,9 @@ void testNoBlockingForReactiveThreads() {
 		IntegrationFlow flow =
 				f -> f
 						.handle(RSockets.outboundGateway("/lowercase")
-								.clientRSocketConnector(this.clientRSocketConnector))
+										.clientRSocketConnector(this.clientRSocketConnector),
+								endpoint -> endpoint.customizeMonoReply((message, mono) ->
+										mono.publishOn(Schedulers.boundedElastic())))
 						.transform("{ firstResult: payload }")
 						.enrich(e -> e
 								.requestPayloadExpression("payload.firstResult")