Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make RSocketOutboundGateway for server side

Browse Source 
* 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.
This commit is contained in:
Artem Bilan
2019-04-26 09:14:42 -04:00
committed by Gary Russell
parent b069780381
commit 4199ff57cd
6 changed files with 470 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java Normal file
View File

@@ -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<RSocketFactory.ClientRSocketFactory> factoryConfigurer = (clientRSocketFactory) -> { };
private Mono<RSocket> 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<RSocketFactory.ClientRSocketFactory> 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<RSocketRequester> getRSocketRequester() {
return this.rsocketMono
.map(rsocket -> RSocketRequester.wrap(rsocket, this.dataMimeType, this.rsocketStrategies))
.cache();
}
@Override
public void destroy() {
this.rsocketMono
.doOnNext(Disposable::dispose)
.subscribe();
}
}

4
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes for RSocket XML namespace parsing and configuration support.
*/
package org.springframework.integration.rsocket.config;

4
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes representing inbound RSocket components.
*/
package org.springframework.integration.rsocket.inbound;

125
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java
View File

@@ -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.
* <p>
* 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.
* <p>
* 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.
* <p>
* 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<RSocketFactory.ClientRSocketFactory> factoryConfigurer = (clientRSocketFactory) -> { };
private Consumer<RSocketStrategies.Builder> 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<RSocketRequester> rSocketRequesterMono;
private EvaluationContext evaluationContext;
public RSocketOutboundGateway(ClientTransport clientTransport, String route) {
this(clientTransport, new ValueExpression<>(route));
@Nullable
private Mono<RSocketRequester> 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<RSocketFactory.ClientRSocketFactory> factoryConfigurer) {
Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null");
this.factoryConfigurer = factoryConfigurer;
}
public void setStrategiesConfigurer(Consumer<RSocketStrategies.Builder> 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<RSocketRequester> 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));

4
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides common classes for RSocket components.
*/
package org.springframework.integration.rsocket;

317
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java
View File

@@ -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<Message<?>>((m) -> m.getHeaders().get(ROUTE_HEADER)));
new RSocketOutboundGateway(
new FunctionExpression<Message<?>>((m) ->
m.getHeaders().get(ROUTE_HEADER)));
rsocketOutboundGateway.setCommandExpression(
new FunctionExpression<Message<?>>((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<String> fireForgetPayloads = ReplayProcessor.create();
final MonoProcessor<RSocketRequester> 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);
}
}
}