Upgrade to Rsocket 1.0-RC7; fix deprecations
* Fix RSocket docs according new functionality **Cherry-pick to 5.2.x**
This commit is contained in:
committed by
Gary Russell
parent
2d9a5f60f4
commit
9ec6529992
@@ -89,14 +89,14 @@ ext {
|
||||
reactorVersion = 'Dysprosium-SR6'
|
||||
resilience4jVersion = '1.3.1'
|
||||
romeToolsVersion = '1.12.2'
|
||||
rsocketVersion = '1.0.0-RC6'
|
||||
rsocketVersion = '1.0.0-RC7'
|
||||
servletApiVersion = '4.0.1'
|
||||
smackVersion = '4.3.4'
|
||||
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.2.5.RELEASE'
|
||||
springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : 'Neumann-RC1'
|
||||
springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '5.3.1.RELEASE'
|
||||
springRetryVersion = '1.2.5.RELEASE'
|
||||
springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.5.RELEASE'
|
||||
springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.6.RELEASE'
|
||||
springWsVersion = '3.0.8.RELEASE'
|
||||
tomcatVersion = "9.0.33"
|
||||
xstreamVersion = '1.4.11.1'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2019 the original author or authors.
|
||||
* Copyright 2019-2020 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.
|
||||
@@ -21,11 +21,12 @@ import java.util.Arrays;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.messaging.rsocket.ClientRSocketFactoryConfigurer;
|
||||
import org.springframework.messaging.rsocket.RSocketConnectorConfigurer;
|
||||
import org.springframework.messaging.rsocket.RSocketRequester;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.MimeType;
|
||||
|
||||
import io.rsocket.core.RSocketConnector;
|
||||
import io.rsocket.transport.ClientTransport;
|
||||
import io.rsocket.transport.netty.client.TcpClientTransport;
|
||||
import io.rsocket.transport.netty.client.WebsocketClientTransport;
|
||||
@@ -48,7 +49,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
|
||||
|
||||
private final Map<Object, MimeType> setupMetadata = new LinkedHashMap<>(4);
|
||||
|
||||
private ClientRSocketFactoryConfigurer factoryConfigurer = (clientRSocketFactory) -> { };
|
||||
private RSocketConnectorConfigurer connectorConfigurer = (connector) -> { };
|
||||
|
||||
private Object setupData;
|
||||
|
||||
@@ -92,18 +93,40 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
|
||||
|
||||
/**
|
||||
* Callback to configure the {@code ClientRSocketFactory} directly.
|
||||
* Note: this class adds extra {@link ClientRSocketFactoryConfigurer} to the
|
||||
* Note: this class adds extra {@link org.springframework.messaging.rsocket.ClientRSocketFactoryConfigurer} to the
|
||||
* target {@link RSocketRequester} to populate a reference to an internal
|
||||
* {@link IntegrationRSocketMessageHandler#responder()}.
|
||||
* This overrides possible external
|
||||
* {@link io.rsocket.RSocketFactory.ClientRSocketFactory#acceptor(io.rsocket.SocketAcceptor)}
|
||||
* @param factoryConfigurer the {@link ClientRSocketFactoryConfigurer} to
|
||||
* @param factoryConfigurer the {@link org.springframework.messaging.rsocket.ClientRSocketFactoryConfigurer} to
|
||||
* configure the {@link io.rsocket.RSocketFactory.ClientRSocketFactory}.
|
||||
* @see RSocketRequester.Builder#rsocketFactory(ClientRSocketFactoryConfigurer)
|
||||
* @see RSocketRequester.Builder#rsocketFactory(org.springframework.messaging.rsocket.ClientRSocketFactoryConfigurer)
|
||||
* @deprecated since 5.2.6 in favor of {@link #setConnectorConfigurer(RSocketConnectorConfigurer)}
|
||||
*/
|
||||
public void setFactoryConfigurer(ClientRSocketFactoryConfigurer factoryConfigurer) {
|
||||
@Deprecated
|
||||
public void setFactoryConfigurer(
|
||||
org.springframework.messaging.rsocket.ClientRSocketFactoryConfigurer factoryConfigurer) {
|
||||
|
||||
Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null");
|
||||
this.factoryConfigurer = factoryConfigurer;
|
||||
setConnectorConfigurer((connector) ->
|
||||
factoryConfigurer.configure(new io.rsocket.RSocketFactory.ClientRSocketFactory(connector)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Callback to configure the {@code ClientRSocketFactory} directly.
|
||||
* Note: this class adds extra {@link RSocketConnectorConfigurer} to the
|
||||
* target {@link RSocketRequester} to populate a reference to an internal
|
||||
* {@link IntegrationRSocketMessageHandler#responder()}.
|
||||
* This overrides possible external
|
||||
* {@link RSocketConnector#acceptor(io.rsocket.SocketAcceptor)}
|
||||
* @param connectorConfigurer the {@link RSocketConnectorConfigurer} to
|
||||
* configure the {@link RSocketConnector}.
|
||||
* @since 5.2.6
|
||||
* @see RSocketRequester.Builder#rsocketConnector(RSocketConnectorConfigurer)
|
||||
*/
|
||||
public void setConnectorConfigurer(RSocketConnectorConfigurer connectorConfigurer) {
|
||||
Assert.notNull(connectorConfigurer, "'connectorConfigurer' must not be null");
|
||||
this.connectorConfigurer = connectorConfigurer;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -160,9 +183,9 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
|
||||
.rsocketStrategies(getRSocketStrategies())
|
||||
.setupData(this.setupData)
|
||||
.setupRoute(this.setupRoute, this.setupRouteVars)
|
||||
.rsocketFactory(this.factoryConfigurer)
|
||||
.rsocketFactory((rsocketFactory) ->
|
||||
rsocketFactory.acceptor(this.rSocketMessageHandler.responder()))
|
||||
.rsocketConnector(this.connectorConfigurer)
|
||||
.rsocketConnector((connector) ->
|
||||
connector.acceptor(this.rSocketMessageHandler.responder()))
|
||||
.apply((builder) -> this.setupMetadata.forEach(builder::setupMetadata))
|
||||
.connect(this.clientTransport)
|
||||
.cache();
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2019 the original author or authors.
|
||||
* Copyright 2019-2020 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.
|
||||
@@ -31,7 +31,8 @@ import org.springframework.messaging.rsocket.RSocketStrategies;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.MimeType;
|
||||
|
||||
import io.rsocket.RSocketFactory;
|
||||
import io.rsocket.core.RSocketConnector;
|
||||
import io.rsocket.core.RSocketServer;
|
||||
import io.rsocket.transport.ServerTransport;
|
||||
import io.rsocket.transport.netty.server.CloseableChannel;
|
||||
import io.rsocket.transport.netty.server.TcpServerTransport;
|
||||
@@ -47,13 +48,13 @@ import reactor.netty.http.server.HttpServer;
|
||||
*
|
||||
* @since 5.2
|
||||
*
|
||||
* @see RSocketFactory.ServerRSocketFactory
|
||||
* @see RSocketConnector
|
||||
*/
|
||||
public class ServerRSocketConnector extends AbstractRSocketConnector implements ApplicationEventPublisherAware {
|
||||
|
||||
private final ServerTransport<CloseableChannel> serverTransport;
|
||||
|
||||
private Consumer<RSocketFactory.ServerRSocketFactory> factoryConfigurer = (serverRSocketFactory) -> { };
|
||||
private Consumer<RSocketServer> serverConfigurer = (rsocketServer) -> { };
|
||||
|
||||
private Mono<CloseableChannel> serverMono;
|
||||
|
||||
@@ -104,12 +105,24 @@ public class ServerRSocketConnector extends AbstractRSocketConnector implements
|
||||
}
|
||||
|
||||
/**
|
||||
* Provide a {@link Consumer} to configure the {@link RSocketFactory.ServerRSocketFactory}.
|
||||
* @param factoryConfigurer the {@link Consumer} to configure the {@link RSocketFactory.ServerRSocketFactory}.
|
||||
* Provide a {@link Consumer} to configure the {@link io.rsocket.RSocketFactory.ServerRSocketFactory}.
|
||||
* @param factoryConfigurer the {@link Consumer} to configure the {@link io.rsocket.RSocketFactory.ServerRSocketFactory}.
|
||||
* @deprecated since 5.2.6 in favor of {@link #setServerConfigurer(Consumer)}
|
||||
*/
|
||||
public void setFactoryConfigurer(Consumer<RSocketFactory.ServerRSocketFactory> factoryConfigurer) {
|
||||
@Deprecated
|
||||
public void setFactoryConfigurer(Consumer<io.rsocket.RSocketFactory.ServerRSocketFactory> factoryConfigurer) {
|
||||
Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null");
|
||||
this.factoryConfigurer = factoryConfigurer;
|
||||
setServerConfigurer((server) ->
|
||||
factoryConfigurer.accept(new io.rsocket.RSocketFactory.ServerRSocketFactory(server)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Provide a {@link Consumer} to configure the {@link RSocketServer}.
|
||||
* @param serverConfigurer the {@link Consumer} to configure the {@link RSocketServer}.
|
||||
* @since 5.2.6
|
||||
*/
|
||||
public void setServerConfigurer(Consumer<RSocketServer> serverConfigurer) {
|
||||
this.serverConfigurer = serverConfigurer;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -164,14 +177,13 @@ public class ServerRSocketConnector extends AbstractRSocketConnector implements
|
||||
public void afterPropertiesSet() {
|
||||
if (this.serverTransport != null) {
|
||||
super.afterPropertiesSet();
|
||||
RSocketFactory.ServerRSocketFactory serverFactory = RSocketFactory.receive();
|
||||
this.factoryConfigurer.accept(serverFactory);
|
||||
RSocketServer rsocketServer = RSocketServer.create();
|
||||
this.serverConfigurer.accept(rsocketServer);
|
||||
|
||||
this.serverMono =
|
||||
serverFactory
|
||||
rsocketServer
|
||||
.acceptor(serverRSocketMessageHandler().responder())
|
||||
.transport(this.serverTransport)
|
||||
.start()
|
||||
.bind(this.serverTransport)
|
||||
.cache();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2019 the original author or authors.
|
||||
* Copyright 2019-2020 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.
|
||||
@@ -44,7 +44,7 @@ import org.springframework.messaging.rsocket.RSocketRequester;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
|
||||
import io.rsocket.RSocketFactory;
|
||||
import io.rsocket.core.RSocketServer;
|
||||
import io.rsocket.frame.decoder.PayloadDecoder;
|
||||
import io.rsocket.transport.netty.server.CloseableChannel;
|
||||
import io.rsocket.transport.netty.server.TcpServerTransport;
|
||||
@@ -185,11 +185,10 @@ public class RSocketInboundGatewayIntegrationTests {
|
||||
|
||||
@Bean
|
||||
public CloseableChannel rsocketServer() {
|
||||
return RSocketFactory.receive()
|
||||
.frameDecoder(PayloadDecoder.ZERO_COPY)
|
||||
return RSocketServer.create()
|
||||
.payloadDecoder(PayloadDecoder.ZERO_COPY)
|
||||
.acceptor(serverRSocketMessageHandler().responder())
|
||||
.transport(TcpServerTransport.create("localhost", 0))
|
||||
.start()
|
||||
.bind(TcpServerTransport.create("localhost", 0))
|
||||
.block();
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2019 the original author or authors.
|
||||
* Copyright 2019-2020 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.
|
||||
@@ -59,7 +59,7 @@ import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
|
||||
import io.rsocket.RSocket;
|
||||
import io.rsocket.RSocketFactory;
|
||||
import io.rsocket.core.RSocketServer;
|
||||
import io.rsocket.frame.decoder.PayloadDecoder;
|
||||
import io.rsocket.transport.netty.server.CloseableChannel;
|
||||
import io.rsocket.transport.netty.server.TcpServerTransport;
|
||||
@@ -112,11 +112,10 @@ public class RSocketOutboundGatewayIntegrationTests {
|
||||
@BeforeAll
|
||||
static void setup() {
|
||||
serverContext = new AnnotationConfigApplicationContext(ServerConfig.class);
|
||||
server = RSocketFactory.receive()
|
||||
.frameDecoder(PayloadDecoder.ZERO_COPY)
|
||||
server = RSocketServer.create()
|
||||
.payloadDecoder(PayloadDecoder.ZERO_COPY)
|
||||
.acceptor(serverContext.getBean(RSocketMessageHandler.class).responder())
|
||||
.transport(TcpServerTransport.create("localhost", 0))
|
||||
.start()
|
||||
.bind(TcpServerTransport.create("localhost", 0))
|
||||
.block();
|
||||
|
||||
serverController = serverContext.getBean(TestController.class);
|
||||
@@ -453,8 +452,8 @@ public class RSocketOutboundGatewayIntegrationTests {
|
||||
.extracting(Message::getPayload)
|
||||
.isInstanceOf(MessageHandlingException.class)
|
||||
.satisfies((ex) -> assertThat((Exception) ex)
|
||||
.hasMessageContaining("io.rsocket.exceptions.ApplicationErrorException: " +
|
||||
"No handler for destination 'invalid'"));
|
||||
.hasMessageContaining(
|
||||
"ApplicationErrorException (0x201): No handler for destination 'invalid'"));
|
||||
|
||||
disposable.dispose();
|
||||
}
|
||||
@@ -501,7 +500,9 @@ public class RSocketOutboundGatewayIntegrationTests {
|
||||
public RSocket rsocketForServerRequests() {
|
||||
return RSocketRequester.builder()
|
||||
.setupRoute("clientConnect")
|
||||
.rsocketFactory(RSocketMessageHandler.clientResponder(RSocketStrategies.create(), controller()))
|
||||
.rsocketConnector(connector ->
|
||||
connector.acceptor(
|
||||
RSocketMessageHandler.responder(RSocketStrategies.create(), controller())))
|
||||
.connectTcp("localhost", server.address().getPort())
|
||||
.block()
|
||||
.rsocket();
|
||||
|
||||
@@ -30,7 +30,7 @@ Before starting an integration flow processing via channel adapters, we need to
|
||||
For this purpose, Spring Integration RSocket support provides the `ServerRSocketConnector` and `ClientRSocketConnector` implementations of the `AbstractRSocketConnector`.
|
||||
|
||||
The `ServerRSocketConnector` exposes a listener on the host and port according to provided `io.rsocket.transport.ServerTransport` for accepting connections from clients.
|
||||
An internal `RSocketFactory.ServerRSocketFactory` instance can be customized with the `setFactoryConfigurer()`, as well as other options that can be configured, e.g. `RSocketStrategies` and `MimeType` for payload data and headers metadata.
|
||||
An internal `RSocketServer` instance can be customized with the `setServerConfigurer()`, as well as other options that can be configured, e.g. `RSocketStrategies` and `MimeType` for payload data and headers metadata.
|
||||
When a `setupRoute` is provided from the client requester (see `ClientRSocketConnector` below), a connected client is stored as a `RSocketRequester` under the key determined by the `clientRSocketKeyStrategy` `BiFunction<Map<String, Object>, DataBuffer, Object>`.
|
||||
By default a connect data is used for the key as a converted value to string with UTF-8 charset.
|
||||
Such an `RSocketRequester` registry can be used in the application logic to determine a particular client connection for interaction with it, or for publishing the same message to all connected clients.
|
||||
@@ -58,7 +58,7 @@ public ServerRSocketConnector serverRSocketConnector() {
|
||||
ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector("localhost", 0);
|
||||
serverRSocketConnector.setRSocketStrategies(rsocketStrategies());
|
||||
serverRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0"));
|
||||
serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
|
||||
serverRSocketConnector.setServerConfigurer((server) -> server.payloadDecoder(PayloadDecoder.ZERO_COPY));
|
||||
serverRSocketConnector.setClientRSocketKeyStrategy((headers, data) -> ""
|
||||
+ headers.get(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER));
|
||||
return serverRSocketConnector;
|
||||
@@ -80,7 +80,7 @@ In addition the `ServerRSocketMessageHandler` can be configured with a `messageM
|
||||
This can be useful in mixed configurations, when classic `@MessageMapping` methods are present in the same application along with RSocket channel adapters and an externally configured RSocket server is present in the application.
|
||||
|
||||
The `ClientRSocketConnector` serves as a holder for `RSocketRequester` based on the `RSocket` connected via the provided `ClientTransport`.
|
||||
The `RSocketFactory.ClientRSocketFactory` can be customized with the provided `ClientRSocketFactoryConfigurer`.
|
||||
The `RSocketConnector` can be customized with the provided `RSocketConnectorConfigurer`.
|
||||
The `setupRoute` (with optional templates variables) and `setupData` with metadata can be also configured on this component.
|
||||
|
||||
A typical client configuration might look like this:
|
||||
|
||||
Reference in New Issue
Block a user