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

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()));
+	}
+
 }

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<String> 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<String> fireForgetPayloads = Sinks.many().replay().all();
 
+		final Sinks.Many<String> voidChannelPayloads = Sinks.many().replay().all();
+
 		final Sinks.One<RSocketRequester> 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<Void> voidChannel(Flux<String> payloads) {
+			return payloads.map(voidChannelPayloads::tryEmitNext).then();
+		}
+
 		@MessageMapping("thrown-exception")
 		Mono<String> handleAndThrow(String payload) {
 			throw new IllegalArgumentException("Invalid input error");

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 <<rsocket-java-config>> for samples how to configure an `RSocketOutboundGateway` endpoint a deal with payloads downstream.
 
 [[rsocket-namespace]]