From 4199ff57cd36a593873a90d92fabffc2eae1f953 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 26 Apr 2019 09:14:42 -0400 Subject: [PATCH] Make `RSocketOutboundGateway` for server side * Introduce a `ClientRSocketConnector` to represent a common logic for client side connection and management an obtained `RSocket` * Rework the `RSocketOutboundGateway` to perform similar RSocket request logic on the server side as well. * Refactor `RSocketOutboundGatewayIntegrationTests` to demonstrate that RSocket requests work the same way from the server side as well. For this reason a `CommonConfig` has been extracted with the same `RSocketOutboundGateway` configuration reuse on both server and client sides. --- .../rsocket/ClientRSocketConnector.java | 124 +++++++ .../rsocket/config/package-info.java | 4 + .../rsocket/inbound/package-info.java | 4 + .../outbound/RSocketOutboundGateway.java | 125 ++++--- .../integration/rsocket/package-info.java | 4 + ...SocketOutboundGatewayIntegrationTests.java | 317 ++++++++++++++---- 6 files changed, 470 insertions(+), 108 deletions(-) create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java new file mode 100644 index 0000000000..b4002d7eb3 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java @@ -0,0 +1,124 @@ +/* + * Copyright 2019 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.rsocket; + +import java.net.URI; +import java.util.function.Consumer; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; + +import io.rsocket.Payload; +import io.rsocket.RSocket; +import io.rsocket.RSocketFactory; +import io.rsocket.transport.ClientTransport; +import io.rsocket.transport.netty.client.TcpClientTransport; +import io.rsocket.transport.netty.client.WebsocketClientTransport; +import io.rsocket.util.DefaultPayload; +import io.rsocket.util.EmptyPayload; +import reactor.core.Disposable; +import reactor.core.publisher.Mono; + +/** + * A client connector to the RSocket server. + * + * @author Artem Bilan + * + * @since 5.2 + * + * @see RSocketFactory.ClientRSocketFactory + * @see RSocketRequester + */ +public class ClientRSocketConnector implements InitializingBean, DisposableBean { + + private final ClientTransport clientTransport; + + private MimeType dataMimeType = MimeTypeUtils.TEXT_PLAIN; + + private Payload connectPayload = EmptyPayload.INSTANCE; + + private RSocketStrategies rsocketStrategies = RSocketStrategies.builder().build(); + + private Consumer factoryConfigurer = (clientRSocketFactory) -> { }; + + private Mono rsocketMono; + + public ClientRSocketConnector(String host, int port) { + this(TcpClientTransport.create(host, port)); + } + + public ClientRSocketConnector(URI uri) { + this(WebsocketClientTransport.create(uri)); + } + + public ClientRSocketConnector(ClientTransport clientTransport) { + Assert.notNull(clientTransport, "'clientTransport' must not be null"); + this.clientTransport = clientTransport; + } + + public void setDataMimeType(MimeType dataMimeType) { + Assert.notNull(dataMimeType, "'dataMimeType' must not be null"); + this.dataMimeType = dataMimeType; + } + + public void setFactoryConfigurer(Consumer factoryConfigurer) { + Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null"); + this.factoryConfigurer = factoryConfigurer; + } + + public void setRSocketStrategies(RSocketStrategies rsocketStrategies) { + Assert.notNull(rsocketStrategies, "'rsocketStrategies' must not be null"); + this.rsocketStrategies = rsocketStrategies; + } + + public void setConnectRoute(String connectRoute) { + this.connectPayload = DefaultPayload.create("", connectRoute); + } + + @Override + public void afterPropertiesSet() { + RSocketFactory.ClientRSocketFactory clientFactory = + RSocketFactory.connect() + .dataMimeType(this.dataMimeType.toString()); + this.factoryConfigurer.accept(clientFactory); + clientFactory.setupPayload(this.connectPayload); + this.rsocketMono = clientFactory.transport(this.clientTransport).start(); + } + + public void connect() { + this.rsocketMono.cache().subscribe(); + } + + public Mono getRSocketRequester() { + return this.rsocketMono + .map(rsocket -> RSocketRequester.wrap(rsocket, this.dataMimeType, this.rsocketStrategies)) + .cache(); + } + + @Override + public void destroy() { + this.rsocketMono + .doOnNext(Disposable::dispose) + .subscribe(); + } + +} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java new file mode 100644 index 0000000000..8dcdba113b --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for RSocket XML namespace parsing and configuration support. + */ +package org.springframework.integration.rsocket.config; diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java new file mode 100644 index 0000000000..b87d225241 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes representing inbound RSocket components. + */ +package org.springframework.integration.rsocket.inbound; 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 c79a7890c5..cc5737ff63 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 @@ -16,8 +16,6 @@ package org.springframework.integration.rsocket.outbound; -import java.util.function.Consumer; - import org.reactivestreams.Publisher; import org.springframework.core.ParameterizedTypeReference; @@ -26,39 +24,51 @@ import org.springframework.expression.Expression; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.rsocket.ClientRSocketConnector; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.rsocket.RSocketRequester; -import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.messaging.rsocket.RSocketRequesterMethodArgumentResolver; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; -import org.springframework.util.MimeType; -import org.springframework.util.MimeTypeUtils; -import io.rsocket.RSocketFactory; -import io.rsocket.transport.ClientTransport; -import reactor.core.Disposable; import reactor.core.publisher.Mono; /** - * An Outbound Messaging Gateway for RSocket client requests. + * An Outbound Messaging Gateway for RSocket requests. + * The request logic is fully based on the {@link RSocketRequester}, which can be obtained from the + * {@link ClientRSocketConnector} on the client side or from the + * {@link RSocketRequesterMethodArgumentResolver#RSOCKET_REQUESTER_HEADER} request message header + * on the server side. + *

+ * An RSocket operation is determined by the configured {@link Command} or respective SpEL + * expression to be evaluated at runtime against the request message. + * By default the {@link Command#requestResponse} operation is used. + *

+ * For a {@link Publisher}-based requests, it must be present in the request message {@code payload}. + * The flattening via upstream {@link org.springframework.integration.channel.FluxMessageChannel} will work, too, + * but this way we will lose a scope of particular request and every {@link Publisher} event + * will be send in its own plain request. + *

+ * If reply is a {@link reactor.core.publisher.Flux}, it is wrapped to the {@link Mono} to retain a request scope. + * The downstream flow is responsible to obtain this {@link reactor.core.publisher.Flux} from a message payload + * and subscribe to it by itself. The {@link Mono} reply from this component is subscribed from the downstream + * {@link org.springframework.integration.channel.FluxMessageChannel} or it is adapted to the + * {@link org.springframework.util.concurrent.ListenableFuture} otherwise. * * @author Artem Bilan * * @since 5.2 * + * @see Command * @see RSocketRequester */ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler { - private final ClientTransport clientTransport; - private final Expression routeExpression; - private MimeType dataMimeType = MimeTypeUtils.TEXT_PLAIN; - - private Consumer factoryConfigurer = (clientRSocketFactory) -> { }; - - private Consumer strategiesConfigurer = (builder) -> { }; + @Nullable + private ClientRSocketConnector clientRSocketConnector; private Expression commandExpression = new ValueExpression<>(Command.requestResponse); @@ -66,42 +76,56 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler private Expression expectedResponseTypeExpression = new ValueExpression<>(String.class); - private Mono rSocketRequesterMono; - private EvaluationContext evaluationContext; - public RSocketOutboundGateway(ClientTransport clientTransport, String route) { - this(clientTransport, new ValueExpression<>(route)); + @Nullable + private Mono rsocketRequesterMono; + + /** + * Instantiate based on the provided RSocket endpoint {@code route}. + * @param route the RSocket endpoint route to use. + */ + public RSocketOutboundGateway(String route) { + this(new ValueExpression<>(route)); } - public RSocketOutboundGateway(ClientTransport clientTransport, Expression routeExpression) { - Assert.notNull(clientTransport, "'clientTransport' must not be null"); + /** + * Instantiate based on the provided SpEL expression to evaluate an RSocket endpoint {@code route} + * at runtime against a request message. + * @param routeExpression the SpEL expression to use. + */ + public RSocketOutboundGateway(Expression routeExpression) { Assert.notNull(routeExpression, "'routeExpression' must not be null"); - this.clientTransport = clientTransport; this.routeExpression = routeExpression; setAsync(true); setPrimaryExpression(this.routeExpression); } - public void setDataMimeType(MimeType dataMimeType) { - Assert.notNull(dataMimeType, "'dataMimeType' must not be null"); - this.dataMimeType = dataMimeType; - } - - public void setFactoryConfigurer(Consumer factoryConfigurer) { - Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null"); - this.factoryConfigurer = factoryConfigurer; - } - - public void setStrategiesConfigurer(Consumer strategiesConfigurer) { - Assert.notNull(strategiesConfigurer, "'strategiesConfigurer' must not be null"); - this.strategiesConfigurer = strategiesConfigurer; + /** + * Configure a {@link ClientRSocketConnector} for client side requests based on the connection + * provided by the {@link ClientRSocketConnector#getRSocketRequester()}. + * In case of server side, an {@link RSocketRequester} must be provided in the + * {@link RSocketRequesterMethodArgumentResolver#RSOCKET_REQUESTER_HEADER} header of request message. + * @param clientRSocketConnector the {@link ClientRSocketConnector} to use. + */ + public void setClientRSocketConnector(ClientRSocketConnector clientRSocketConnector) { + Assert.notNull(clientRSocketConnector, "'clientRSocketConnector' must not be null"); + this.clientRSocketConnector = clientRSocketConnector; } + /** + * Configure a {@link Command} for RSocket request type. + * @param command the {@link Command} to use. + */ public void setCommand(Command command) { setCommandExpression(new ValueExpression<>(command)); } + /** + * Configure a SpEL expression to evaluate a {@link Command} for RSocket request type at runtime + * against a request message. + * @param commandExpression the SpEL expression to use. + */ public void setCommandExpression(Expression commandExpression) { Assert.notNull(commandExpression, "'commandExpression' must not be null"); this.commandExpression = commandExpression; @@ -155,26 +179,35 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler @Override protected void doInit() { super.doInit(); - this.rSocketRequesterMono = - RSocketRequester.builder() - .rsocketFactory(this.factoryConfigurer) - .rsocketStrategies(this.strategiesConfigurer) - .connect(this.clientTransport, this.dataMimeType); - this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + if (this.clientRSocketConnector != null) { + this.rsocketRequesterMono = this.clientRSocketConnector.getRSocketRequester().cache(); + } } @Override public void destroy() { super.destroy(); - this.rSocketRequesterMono.map(RSocketRequester::rsocket) - .doOnNext(Disposable::dispose) - .subscribe(); + } @Override protected Object handleRequestMessage(Message requestMessage) { - return this.rSocketRequesterMono.cache() + RSocketRequester rsocketRequester = requestMessage.getHeaders() + .get(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, RSocketRequester.class); + Mono requesterMono; + if (rsocketRequester != null) { + requesterMono = Mono.just(rsocketRequester); + } + else { + requesterMono = this.rsocketRequesterMono; + } + + Assert.notNull(requesterMono, () -> + "The 'RSocketRequester' must be configured via 'ClientRSocketConnector' or provided in the '" + + RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER + "' request message headers."); + + return requesterMono .map((rSocketRequester) -> createRequestSpec(rSocketRequester, requestMessage)) .map((requestSpec) -> createResponseSpec(requestSpec, requestMessage)) .flatMap((responseSpec) -> performRequest(responseSpec, requestMessage)); diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java new file mode 100644 index 0000000000..1ab963c7af --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides common classes for RSocket components. + */ +package org.springframework.integration.rsocket; 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 99d1539fa7..53e6720a1e 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 @@ -19,10 +19,13 @@ package org.springframework.integration.rsocket.outbound; import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; +import java.util.Collections; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInfo; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.AnnotationConfigApplicationContext; @@ -38,14 +41,17 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.MessageChannels; import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.rsocket.ClientRSocketConnector; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.handler.annotation.MessageExceptionHandler; import org.springframework.messaging.handler.annotation.MessageMapping; import org.springframework.messaging.rsocket.MessageHandlerAcceptor; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketRequesterMethodArgumentResolver; import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.messaging.support.ErrorMessage; import org.springframework.stereotype.Controller; @@ -55,12 +61,12 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import io.netty.buffer.PooledByteBufAllocator; import io.rsocket.RSocketFactory; import io.rsocket.frame.decoder.PayloadDecoder; -import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.server.CloseableChannel; import io.rsocket.transport.netty.server.TcpServerTransport; import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.publisher.MonoProcessor; import reactor.core.publisher.ReplayProcessor; import reactor.netty.tcp.TcpServer; import reactor.test.StepVerifier; @@ -78,12 +84,20 @@ public class RSocketOutboundGatewayIntegrationTests { private static final String COMMAND_HEADER = "rsocket_command"; - private static AnnotationConfigApplicationContext context; + private static AnnotationConfigApplicationContext serverContext; private static int port; private static CloseableChannel server; + private static FluxMessageChannel serverInputChannel; + + private static FluxMessageChannel serverResultChannel; + + private static PollableChannel serverErrorChannel; + + private static TestController serverController; + @Autowired private FluxMessageChannel inputChannel; @@ -93,36 +107,69 @@ public class RSocketOutboundGatewayIntegrationTests { @Autowired private PollableChannel errorChannel; + @Autowired + private TestController clientController; + + @Autowired + private ClientRSocketConnector clientRSocketConnector; + + private RSocketRequester serverRsocketRequester; + @BeforeAll static void setup() { - context = new AnnotationConfigApplicationContext(ServerConfig.class); + serverContext = new AnnotationConfigApplicationContext(ServerConfig.class); TcpServer tcpServer = TcpServer.create().port(0) .doOnBound(server -> port = server.port()); server = RSocketFactory.receive() .frameDecoder(PayloadDecoder.ZERO_COPY) - .acceptor(context.getBean(MessageHandlerAcceptor.class)) + .acceptor(serverContext.getBean(MessageHandlerAcceptor.class)) .transport(TcpServerTransport.create(tcpServer)) .start() .block(); + + serverController = serverContext.getBean(TestController.class); + serverInputChannel = serverContext.getBean("inputChannel", FluxMessageChannel.class); + serverResultChannel = serverContext.getBean("resultChannel", FluxMessageChannel.class); + serverErrorChannel = serverContext.getBean("errorChannel", PollableChannel.class); } @AfterAll static void tearDown() { - context.close(); + serverContext.close(); server.dispose(); } + @BeforeEach + void setupTest(TestInfo testInfo) { + if (testInfo.getDisplayName().startsWith("server")) { + this.clientRSocketConnector.connect(); + this.serverRsocketRequester = serverController.clientRequester.block(Duration.ofSeconds(10)); + } + } + @Test - void fireAndForget() { - Disposable disposable = Flux.from(this.resultChannel).subscribe(); - this.inputChannel.send( + void clientFireAndForget() { + fireAndForget(this.inputChannel, this.resultChannel, serverController, null); + } + + @Test + void serverFireAndForget() { + fireAndForget(serverInputChannel, serverResultChannel, this.clientController, this.serverRsocketRequester); + } + + private void fireAndForget(MessageChannel inputChannel, FluxMessageChannel resultChannel, + TestController controller, RSocketRequester rsocketRequester) { + + Disposable disposable = Flux.from(resultChannel).subscribe(); + inputChannel.send( MessageBuilder.withPayload("Hello") .setHeader(ROUTE_HEADER, "receive") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.fireAndForget) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); - StepVerifier.create(context.getBean(ServerController.class).fireForgetPayloads) + StepVerifier.create(controller.fireForgetPayloads) .expectNext("Hello") .thenCancel() .verify(); @@ -130,16 +177,29 @@ public class RSocketOutboundGatewayIntegrationTests { disposable.dispose(); } + @Test - void echo() { - this.inputChannel.send( + void clientEcho() { + echo(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverEcho() { + echo(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void echo(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload("Hello") .setHeader(ROUTE_HEADER, "echo") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); StepVerifier.create( - Flux.from(this.resultChannel) + Flux.from(resultChannel) .map(Message::getPayload) .cast(String.class)) .expectNext("Hello") @@ -148,15 +208,27 @@ public class RSocketOutboundGatewayIntegrationTests { } @Test - void echoAsync() { - this.inputChannel.send( + void clientEchoAsync() { + echoAsync(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverEchoAsync() { + echoAsync(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void echoAsync(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload("Hello") .setHeader(ROUTE_HEADER, "echo-async") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); StepVerifier.create( - Flux.from(this.resultChannel) + Flux.from(resultChannel) .map(Message::getPayload) .cast(String.class)) .expectNext("Hello async") @@ -165,15 +237,27 @@ public class RSocketOutboundGatewayIntegrationTests { } @Test - void echoStream() { - this.inputChannel.send( + void clientEchoStream() { + echoStream(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverEchoStream() { + echoStream(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void echoStream(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload("Hello") .setHeader(ROUTE_HEADER, "echo-stream") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); Message resultMessage = - Flux.from(this.resultChannel) + Flux.from(resultChannel) .blockFirst(); assertThat(resultMessage) @@ -191,15 +275,27 @@ public class RSocketOutboundGatewayIntegrationTests { } @Test - void echoChannel() { - this.inputChannel.send( + void clientEchoChannel() { + echoChannel(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverEchoChannel() { + echoChannel(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void echoChannel(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload(Flux.range(1, 10).map(i -> "Hello " + i)) .setHeader(ROUTE_HEADER, "echo-channel") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); Message resultMessage = - Flux.from(this.resultChannel) + Flux.from(resultChannel) .blockFirst(); assertThat(resultMessage) @@ -215,16 +311,29 @@ public class RSocketOutboundGatewayIntegrationTests { .verify(); } + @Test - void voidReturnValue() { - this.inputChannel.send( + void clientVoidReturnValue() { + voidReturnValue(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverVoidReturnValue() { + voidReturnValue(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void voidReturnValue(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload("Hello") .setHeader(ROUTE_HEADER, "void-return-value") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); Message resultMessage = - Flux.from(this.resultChannel) + Flux.from(resultChannel) .blockFirst(); assertThat(resultMessage) @@ -239,15 +348,27 @@ public class RSocketOutboundGatewayIntegrationTests { } @Test - void voidReturnValueFromExceptionHandler() { - this.inputChannel.send( + void clientVoidReturnValueFromExceptionHandler() { + voidReturnValueFromExceptionHandler(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverVoidReturnValueFromExceptionHandler() { + voidReturnValueFromExceptionHandler(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void voidReturnValueFromExceptionHandler(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload("bad") .setHeader(ROUTE_HEADER, "void-return-value") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); Message resultMessage = - Flux.from(this.resultChannel) + Flux.from(resultChannel) .blockFirst(); assertThat(resultMessage) @@ -262,15 +383,27 @@ public class RSocketOutboundGatewayIntegrationTests { } @Test - void handleWithThrownException() { - this.inputChannel.send( + void clientHandleWithThrownException() { + handleWithThrownException(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverHandleWithThrownException() { + handleWithThrownException(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void handleWithThrownException(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload("a") .setHeader(ROUTE_HEADER, "thrown-exception") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); StepVerifier.create( - Flux.from(this.resultChannel) + Flux.from(resultChannel) .map(Message::getPayload) .cast(String.class)) .expectNext("Invalid input error handled") @@ -279,15 +412,27 @@ public class RSocketOutboundGatewayIntegrationTests { } @Test - void handleWithErrorSignal() { - this.inputChannel.send( + void clientHandleWithErrorSignal() { + handleWithErrorSignal(this.inputChannel, this.resultChannel, null); + } + + @Test + void serverHandleWithErrorSignal() { + handleWithErrorSignal(serverInputChannel, serverResultChannel, this.serverRsocketRequester); + } + + private void handleWithErrorSignal(MessageChannel inputChannel, FluxMessageChannel resultChannel, + RSocketRequester rsocketRequester) { + + inputChannel.send( MessageBuilder.withPayload("a") .setHeader(ROUTE_HEADER, "error-signal") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); StepVerifier.create( - Flux.from(this.resultChannel) + Flux.from(resultChannel) .map(Message::getPayload) .cast(String.class)) .expectNext("Invalid input error handled") @@ -296,12 +441,24 @@ public class RSocketOutboundGatewayIntegrationTests { } @Test - void noMatchingRoute() { - Disposable disposable = Flux.from(this.resultChannel).subscribe(); - this.inputChannel.send( + void clientNoMatchingRoute() { + noMatchingRoute(this.inputChannel, this.resultChannel, this.errorChannel, null); + } + + @Test + void serverNoMatchingRoute() { + noMatchingRoute(serverInputChannel, serverResultChannel, serverErrorChannel, this.serverRsocketRequester); + } + + private void noMatchingRoute(MessageChannel inputChannel, FluxMessageChannel resultChannel, + PollableChannel errorChannel, RSocketRequester rsocketRequester) { + + Disposable disposable = Flux.from(resultChannel).subscribe(); + inputChannel.send( MessageBuilder.withPayload("anything") .setHeader(ROUTE_HEADER, "invalid") .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester) .build()); Message errorMessage = errorChannel.receive(10_000); @@ -317,22 +474,30 @@ public class RSocketOutboundGatewayIntegrationTests { disposable.dispose(); } - @Configuration - @EnableIntegration - public static class ClientConfig { + private abstract static class CommonConfig { @Bean - public MessageHandler rsocketOutboundGateway() { + public RSocketStrategies rsocketStrategies() { + return RSocketStrategies.builder() + .decoder(StringDecoder.allMimeTypes()) + .encoder(CharSequenceEncoder.allMimeTypes()) + .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) + .build(); + } + + @Bean + public TestController controller() { + return new TestController(); + } + + @Bean + public RSocketOutboundGateway rsocketOutboundGateway() { RSocketOutboundGateway rsocketOutboundGateway = - new RSocketOutboundGateway(TcpClientTransport.create(port), - new FunctionExpression>((m) -> m.getHeaders().get(ROUTE_HEADER))); + new RSocketOutboundGateway( + new FunctionExpression>((m) -> + m.getHeaders().get(ROUTE_HEADER))); rsocketOutboundGateway.setCommandExpression( new FunctionExpression>((m) -> m.getHeaders().get(COMMAND_HEADER))); - rsocketOutboundGateway.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); - rsocketOutboundGateway.setStrategiesConfigurer((strategies) -> - strategies.decoder(StringDecoder.allMimeTypes()) - .encoder(CharSequenceEncoder.allMimeTypes()) - .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT))); return rsocketOutboundGateway; } @@ -352,36 +517,59 @@ public class RSocketOutboundGatewayIntegrationTests { } @Configuration - static class ServerConfig { + @EnableIntegration + public static class ClientConfig extends CommonConfig { @Bean - public ServerController controller() { - return new ServerController(); - } - - @Bean - public MessageHandlerAcceptor messageHandlerAcceptor() { + public MessageHandlerAcceptor clientAcceptor() { MessageHandlerAcceptor acceptor = new MessageHandlerAcceptor(); + acceptor.setHandlers(Collections.singletonList(controller())); + acceptor.setAutoDetectDisabled(); acceptor.setRSocketStrategies(rsocketStrategies()); return acceptor; } @Bean - public RSocketStrategies rsocketStrategies() { - return RSocketStrategies.builder() - .decoder(StringDecoder.allMimeTypes()) - .encoder(CharSequenceEncoder.allMimeTypes()) - .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) - .build(); + public ClientRSocketConnector clientRSocketConnector() { + ClientRSocketConnector clientRSocketConnector = new ClientRSocketConnector("localhost", port); + clientRSocketConnector.setFactoryConfigurer((factory) -> factory + .frameDecoder(PayloadDecoder.ZERO_COPY) + .acceptor(clientAcceptor())); + clientRSocketConnector.setRSocketStrategies(rsocketStrategies()); + clientRSocketConnector.setConnectRoute("clientConnect"); + return clientRSocketConnector; + } + + @Override + @Bean + public RSocketOutboundGateway rsocketOutboundGateway() { + RSocketOutboundGateway rsocketOutboundGateway = super.rsocketOutboundGateway(); + rsocketOutboundGateway.setClientRSocketConnector(clientRSocketConnector()); + return rsocketOutboundGateway; + } + + } + + @Configuration + @EnableIntegration + static class ServerConfig extends CommonConfig { + + @Bean + public MessageHandlerAcceptor serverAcceptor() { + MessageHandlerAcceptor acceptor = new MessageHandlerAcceptor(); + acceptor.setRSocketStrategies(rsocketStrategies()); + return acceptor; } } @Controller - static class ServerController { + static class TestController { final ReplayProcessor fireForgetPayloads = ReplayProcessor.create(); + final MonoProcessor clientRequester = MonoProcessor.create(); + @MessageMapping("receive") void receive(String payload) { this.fireForgetPayloads.onNext(payload); @@ -434,6 +622,11 @@ public class RSocketOutboundGatewayIntegrationTests { return Mono.delay(Duration.ofMillis(10)).then(Mono.empty()); } + @MessageMapping("clientConnect") + void clientConnect(RSocketRequester requester) { + this.clientRequester.onNext(requester); + } + } }