From 9ec652999283aa30cdb611995735ed416dbf9af9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 28 Apr 2020 16:54:01 -0400 Subject: [PATCH] Upgrade to Rsocket 1.0-RC7; fix deprecations * Fix RSocket docs according new functionality **Cherry-pick to 5.2.x** --- build.gradle | 4 +- .../rsocket/ClientRSocketConnector.java | 45 ++++++++++++++----- .../rsocket/ServerRSocketConnector.java | 38 ++++++++++------ ...RSocketInboundGatewayIntegrationTests.java | 11 +++-- ...SocketOutboundGatewayIntegrationTests.java | 19 ++++---- src/reference/asciidoc/rsocket.adoc | 6 +-- 6 files changed, 79 insertions(+), 44 deletions(-) diff --git a/build.gradle b/build.gradle index 3987b682e8..fd35364c19 100644 --- a/build.gradle +++ b/build.gradle @@ -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' 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 index 188dbd4f0c..998a168b05 100644 --- 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 @@ -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 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(); diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java index 20640195ab..4cfdc33eb1 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java @@ -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 serverTransport; - private Consumer factoryConfigurer = (serverRSocketFactory) -> { }; + private Consumer serverConfigurer = (rsocketServer) -> { }; private Mono 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 factoryConfigurer) { + @Deprecated + public void setFactoryConfigurer(Consumer 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 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(); } } diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java index f740ec93da..04b1a029f4 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java @@ -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(); } 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 f8c07b9bf7..a90df41a8f 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 @@ -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(); diff --git a/src/reference/asciidoc/rsocket.adoc b/src/reference/asciidoc/rsocket.adoc index b78038a1eb..d10a6d8b49 100644 --- a/src/reference/asciidoc/rsocket.adoc +++ b/src/reference/asciidoc/rsocket.adoc @@ -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, 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: