From 2ed6d3e301f1ea02eb137247e73ac60f45e48b1f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 7 Oct 2024 14:51:49 -0400 Subject: [PATCH] GH-9538: Rely on the `customizeMonoReply()` for thread switching Fixes: https://github.com/spring-projects/spring-integration/issues/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()))) ``` --- .../handler/AbstractMessageProducingHandler.java | 2 -- .../integration/rsocket/dsl/RSocketDslTests.java | 6 ++++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 3a03e66acb..5f72bd7e7b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -33,7 +33,6 @@ import org.reactivestreams.Publisher; 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 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan CompletableFuture 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 diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java index aed3d7dd5b..978bd10044 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java @@ -21,6 +21,7 @@ import java.util.function.Function; 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 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Artem Bilan - * * @since 5.2 */ @SpringJUnitConfig @@ -86,7 +86,9 @@ public class RSocketDslTests { 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")