From fa1ab223aec1a716dc2fdcf80e8c95f5748c4402 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 21 Dec 2020 17:18:37 -0500 Subject: [PATCH] Special handle for void in the RSocketOutGateway To avoid extra `Publisher` subscription work in the target application "flatten" a returned empty `Flux` for the `void` expected type into a `Mono.empty()` for automatic subscription on the output channel --- .../outbound/RSocketOutboundGateway.java | 15 +++++++-- ...SocketOutboundGatewayIntegrationTests.java | 31 +++++++++++++++++++ src/reference/asciidoc/rsocket.adoc | 3 ++ 3 files changed, 47 insertions(+), 2 deletions(-) diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java index c9ba953c75..4fb1ba4b4d 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java @@ -22,6 +22,7 @@ import java.util.Map; import org.reactivestreams.Publisher; import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ResolvableType; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.integration.expression.ExpressionUtils; @@ -38,6 +39,7 @@ import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** @@ -298,12 +300,17 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler } case requestStream: case requestChannel: + Flux result; + ResolvableType expectedType; if (expectedResponseType instanceof Class) { - return Mono.just(retrieveSpec.retrieveFlux((Class) expectedResponseType)); + expectedType = ResolvableType.forClass((Class) expectedResponseType); + result = retrieveSpec.retrieveFlux((Class) expectedResponseType); } else { - return Mono.just(retrieveSpec.retrieveFlux((ParameterizedTypeReference) expectedResponseType)); + expectedType = ResolvableType.forType((ParameterizedTypeReference) expectedResponseType); + result = retrieveSpec.retrieveFlux((ParameterizedTypeReference) expectedResponseType); } + return isVoid(expectedType) ? result.then() : Mono.just(result); default: throw new UnsupportedOperationException("Unsupported interaction model: " + interactionModel); } @@ -346,4 +353,8 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler } } + private static boolean isVoid(ResolvableType type) { + return (Void.class.equals(type.resolve()) || void.class.equals(type.resolve())); + } + } diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java index 39e92f4d07..b0d2f3e572 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java @@ -106,6 +106,9 @@ public class RSocketOutboundGatewayIntegrationTests { @Autowired private TestController clientController; + @Autowired + RSocketOutboundGateway clientRsocketOutboundGateway; + private RSocketRequester serverRsocketRequester; @BeforeAll @@ -433,6 +436,27 @@ public class RSocketOutboundGatewayIntegrationTests { noMatchingRoute(serverInputChannel, serverResultChannel, serverErrorChannel, this.serverRsocketRequester); } + @Test + void voidRequestChannel() { + Disposable disposable = Flux.from(resultChannel).subscribe(); + this.clientRsocketOutboundGateway.setExpectedResponseType(void.class); + Flux testData = Flux.range(1, 10).map(i -> "Hello " + i); + this.inputChannel.send( + MessageBuilder.withPayload(testData) + .setHeader(ROUTE_HEADER, "void-channel") + .setHeader(INTERACTION_MODEL_HEADER, RSocketInteractionModel.requestChannel) + .build()); + + StepVerifier.create(serverController.voidChannelPayloads.asFlux()) + .expectNext(testData.toStream().toArray(String[]::new)) + .thenCancel() + .verify(Duration.ofSeconds(10)); + + + disposable.dispose(); + this.clientRsocketOutboundGateway.setExpectedResponseType(String.class); + } + private void noMatchingRoute(MessageChannel inputChannel, FluxMessageChannel resultChannel, PollableChannel errorChannel, RSocketRequester rsocketRequester) { @@ -540,6 +564,8 @@ public class RSocketOutboundGatewayIntegrationTests { final Sinks.Many fireForgetPayloads = Sinks.many().replay().all(); + final Sinks.Many voidChannelPayloads = Sinks.many().replay().all(); + final Sinks.One clientRequester = Sinks.one(); @MessageMapping("receive") @@ -567,6 +593,11 @@ public class RSocketOutboundGatewayIntegrationTests { return payloads.delayElements(Duration.ofMillis(10)).map(payload -> payload + " async"); } + @MessageMapping("void-channel") + Mono voidChannel(Flux payloads) { + return payloads.map(voidChannelPayloads::tryEmitNext).then(); + } + @MessageMapping("thrown-exception") Mono handleAndThrow(String payload) { throw new IllegalArgumentException("Invalid input error"); diff --git a/src/reference/asciidoc/rsocket.adoc b/src/reference/asciidoc/rsocket.adoc index d10a6d8b49..5eb3b65f47 100644 --- a/src/reference/asciidoc/rsocket.adoc +++ b/src/reference/asciidoc/rsocket.adoc @@ -188,6 +188,9 @@ public Flux flattenRSocketResponse(Flux payload) { Or subscribed explicitly in the target application logic. +The expected response type can also be configured (or evaluated via expression) to `void` treating this gateway as an outbound channel adapter. +However the `outputChannel` still has to be configured (even if it just a `NullChannel`) to initiate a subscription to the returned `Mono`. + See <> for samples how to configure an `RSocketOutboundGateway` endpoint a deal with payloads downstream. [[rsocket-namespace]]