From 9fe1feea9a91c5826c07be8fa2f00edb6b5c5107 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Wed, 22 Jul 2020 14:57:33 +0300 Subject: [PATCH] Provide access to underlying RSocketClient The recently introduced support for RSocketClient in commit 7c982511426eebea3d34730ab128bb39a55a9de6 did not expose the underlying client in order to avoid a hard dependency on RSocket 1.1 for the time being. However such access appears to be necessary, e.g. for Spring Integration, where the connection needs to be established (warmed up) ahead of actual requests. See gh-25332 --- .../rsocket/DefaultRSocketRequester.java | 144 +++--------------- .../DefaultRSocketRequesterBuilder.java | 7 +- .../messaging/rsocket/RSocketRequester.java | 29 ++-- .../DefaultRSocketRequesterBuilderTests.java | 4 +- .../rsocket/RSocketBufferLeakTests.java | 2 +- ...RSocketClientToServerIntegrationTests.java | 2 +- ...RSocketServerToClientIntegrationTests.java | 6 +- ...lientToServerCoroutinesIntegrationTests.kt | 2 +- 8 files changed, 43 insertions(+), 153 deletions(-) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java index 9769674e5b..8376f4e026 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java @@ -23,6 +23,7 @@ import java.util.function.Consumer; import io.rsocket.Payload; import io.rsocket.RSocket; import io.rsocket.RSocketClient; +import io.rsocket.core.RSocketClientAdapter; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -49,7 +50,10 @@ final class DefaultRSocketRequester implements RSocketRequester { private static final Map EMPTY_HINTS = Collections.emptyMap(); - private final RSocketDelegate rsocketDelegate; + private final RSocketClient rsocketClient; + + @Nullable + private final RSocket rsocket; private final MimeType dataMimeType; @@ -61,15 +65,16 @@ final class DefaultRSocketRequester implements RSocketRequester { DefaultRSocketRequester( - RSocketDelegate rsocketDelegate, MimeType dataMimeType, MimeType metadataMimeType, - RSocketStrategies strategies) { + @Nullable RSocketClient rsocketClient, @Nullable RSocket rsocket, + MimeType dataMimeType, MimeType metadataMimeType, RSocketStrategies strategies) { - Assert.notNull(rsocketDelegate, "RSocket or RSocketClient is required"); + Assert.isTrue(rsocketClient != null || rsocket != null, "RSocketClient or RSocket is required"); Assert.notNull(dataMimeType, "'dataMimeType' is required"); Assert.notNull(metadataMimeType, "'metadataMimeType' is required"); Assert.notNull(strategies, "RSocketStrategies is required"); - this.rsocketDelegate = rsocketDelegate; + this.rsocketClient = (rsocketClient != null ? rsocketClient : new RSocketClientAdapter(rsocket)); + this.rsocket = rsocket; this.dataMimeType = dataMimeType; this.metadataMimeType = metadataMimeType; this.strategies = strategies; @@ -77,11 +82,15 @@ final class DefaultRSocketRequester implements RSocketRequester { } + @Override + public RSocketClient rsocketClient() { + return this.rsocketClient; + } + @Nullable @Override public RSocket rsocket() { - return (this.rsocketDelegate instanceof ConnectionRSocketDelegate ? - ((ConnectionRSocketDelegate) this.rsocketDelegate).getRSocket() : null); + return this.rsocket; } @Override @@ -104,11 +113,6 @@ final class DefaultRSocketRequester implements RSocketRequester { return new DefaultRequestSpec(metadata, mimeType); } - @Override - public void dispose() { - this.rsocketDelegate.dispose(); - } - private static boolean isVoid(ResolvableType elementType) { return (Void.class.equals(elementType.resolve()) || void.class.equals(elementType.resolve())); } @@ -256,12 +260,12 @@ final class DefaultRSocketRequester implements RSocketRequester { @Override public Mono sendMetadata() { - return rsocketDelegate().metadataPush(getPayloadMono()); + return rsocketClient.metadataPush(getPayloadMono()); } @Override public Mono send() { - return rsocketDelegate().fireAndForget(getPayloadMono()); + return rsocketClient.fireAndForget(getPayloadMono()); } @Override @@ -276,7 +280,7 @@ final class DefaultRSocketRequester implements RSocketRequester { @SuppressWarnings("unchecked") private Mono retrieveMono(ResolvableType elementType) { - Mono payloadMono = rsocketDelegate().requestResponse(getPayloadMono()); + Mono payloadMono = rsocketClient.requestResponse(getPayloadMono()); if (isVoid(elementType)) { return (Mono) payloadMono.then(); @@ -301,8 +305,8 @@ final class DefaultRSocketRequester implements RSocketRequester { private Flux retrieveFlux(ResolvableType elementType) { Flux payloadFlux = (this.payloadFlux != null ? - rsocketDelegate().requestChannel(this.payloadFlux) : - rsocketDelegate().requestStream(getPayloadMono())); + rsocketClient.requestChannel(this.payloadFlux) : + rsocketClient.requestStream(getPayloadMono())); if (isVoid(elementType)) { return payloadFlux.thenMany(Flux.empty()); @@ -313,10 +317,6 @@ final class DefaultRSocketRequester implements RSocketRequester { (T) decoder.decode(dataBuffer, elementType, dataMimeType, EMPTY_HINTS)); } - private RSocketDelegate rsocketDelegate() { - return DefaultRSocketRequester.this.rsocketDelegate; - } - private Mono getPayloadMono() { Assert.state(this.payloadFlux == null, "No RSocket interaction with Flux request and Mono response."); return this.payloadMono != null ? this.payloadMono : firstPayload(emptyBufferMono); @@ -327,106 +327,4 @@ final class DefaultRSocketRequester implements RSocketRequester { } } - - // Contract to avoid a hard dependency on RSocketClient for now. - - interface RSocketDelegate { - - Mono fireAndForget(Mono payloadMono); - - Mono requestResponse(Mono payloadMono); - - Flux requestStream(Mono payloadMono); - - Flux requestChannel(Publisher payloadPublisher); - - Mono metadataPush(Mono payloadMono); - - void dispose(); - } - - static class ConnectionRSocketDelegate implements RSocketDelegate { - - private final RSocket rsocket; - - public ConnectionRSocketDelegate(RSocket rsocket) { - Assert.notNull(rsocket, "RSocket is required"); - this.rsocket = rsocket; - } - - public RSocket getRSocket() { - return this.rsocket; - } - - @Override - public Mono fireAndForget(Mono payloadMono) { - return payloadMono.flatMap(this.rsocket::fireAndForget); - } - - @Override - public Mono requestResponse(Mono payloadMono) { - return payloadMono.flatMap(this.rsocket::requestResponse); - } - - @Override - public Flux requestStream(Mono payloadMono) { - return payloadMono.flatMapMany(this.rsocket::requestStream); - } - - @Override - public Flux requestChannel(Publisher payloadPublisher) { - return this.rsocket.requestChannel(payloadPublisher); - } - - @Override - public Mono metadataPush(Mono payloadMono) { - return payloadMono.flatMap(this.rsocket::metadataPush); - } - - @Override - public void dispose() { - this.rsocket.dispose(); - } - } - - static class ClientRSocketDelegate implements RSocketDelegate { - - private final RSocketClient rsocketClient; - - public ClientRSocketDelegate(RSocketClient rsocketClient) { - Assert.notNull(rsocketClient, "RSocketClient is required"); - this.rsocketClient = rsocketClient; - } - - @Override - public Mono fireAndForget(Mono payloadMono) { - return this.rsocketClient.fireAndForget(payloadMono); - } - - @Override - public Mono requestResponse(Mono payloadMono) { - return this.rsocketClient.requestResponse(payloadMono); - } - - @Override - public Flux requestStream(Mono payloadMono) { - return this.rsocketClient.requestStream(payloadMono); - } - - @Override - public Flux requestChannel(Publisher payloadPublisher) { - return this.rsocketClient.requestChannel(payloadPublisher); - } - - @Override - public Mono metadataPush(Mono payloadMono) { - return this.rsocketClient.metadataPush(payloadMono); - } - - @Override - public void dispose() { - this.rsocketClient.dispose(); - } - } - } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilder.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilder.java index 1ae1051798..d4db41a588 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilder.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilder.java @@ -184,8 +184,7 @@ final class DefaultRSocketRequesterBuilder implements RSocketRequester.Builder { metaMimeType, dataMimeType, setupPayload, strategies); return new DefaultRSocketRequester( - new DefaultRSocketRequester.ClientRSocketDelegate(connector.toRSocketClient(transport)), - dataMimeType, metaMimeType, strategies); + connector.toRSocketClient(transport), null, dataMimeType, metaMimeType, strategies); } @Override @@ -219,9 +218,7 @@ final class DefaultRSocketRequesterBuilder implements RSocketRequester.Builder { metaMimeType, dataMimeType, setupPayload, rsocketStrategies); return connector.connect(transport).map(rsocket -> - new DefaultRSocketRequester( - new DefaultRSocketRequester.ConnectionRSocketDelegate(rsocket), - dataMimeType, metaMimeType, rsocketStrategies)); + new DefaultRSocketRequester(null, rsocket, dataMimeType, metaMimeType, rsocketStrategies)); } private RSocketStrategies getRSocketStrategies() { diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java index 1542aa3b68..4aa54121c4 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java @@ -22,6 +22,7 @@ import java.util.function.Consumer; import io.rsocket.ConnectionSetupPayload; import io.rsocket.Payload; import io.rsocket.RSocket; +import io.rsocket.RSocketClient; import io.rsocket.transport.ClientTransport; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.client.WebsocketClientTransport; @@ -48,13 +49,16 @@ import org.springframework.util.MimeType; public interface RSocketRequester { /** - * This method returns {@code null} unless the the requester was created - * with a "live" RSocket through one of the (now deprecated) builder connect - * methods or via {@link #wrap(RSocket, MimeType, MimeType, RSocketStrategies)} - * which is mainly for internal use in client and server responder - * implementations. Otherwise in the more common case where there is no - * "live" RSocket, the requester delegates to an - * {@link io.rsocket.RSocketClient}. + * Return the underlying {@link RSocketClient} used to make requests with. + * @since 5.3 + */ + RSocketClient rsocketClient(); + + /** + * Return the underlying {@link RSocket} if the requester was created with a + * "live" RSocket via {@link #wrap(RSocket, MimeType, MimeType, RSocketStrategies)} + * or via one of the (deprecated) connect methods on the + * {@code RSocketRequester} builder, or otherwise return {@code null}. */ @Nullable RSocket rsocket(); @@ -103,13 +107,6 @@ public interface RSocketRequester { */ RequestSpec metadata(Object metadata, @Nullable MimeType mimeType); - /** - * Invoke the dispose method on the underlying - * {@link io.rsocket.RSocketClient} or {@link RSocket}. - * @since 5.3 - */ - public void dispose(); - /** * Obtain a builder to create a client {@link RSocketRequester} by connecting * to an RSocket server. @@ -126,9 +123,7 @@ public interface RSocketRequester { RSocket rsocket, MimeType dataMimeType, MimeType metadataMimeType, RSocketStrategies strategies) { - return new DefaultRSocketRequester( - new DefaultRSocketRequester.ConnectionRSocketDelegate(rsocket), - dataMimeType, metadataMimeType, strategies); + return new DefaultRSocketRequester(null, rsocket, dataMimeType, metadataMimeType, strategies); } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilderTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilderTests.java index 84bdb56288..cedee19e87 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilderTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilderTests.java @@ -206,8 +206,8 @@ public class DefaultRSocketRequesterBuilderTests { } private ConnectionSetupPayload getConnectionSetupPayload(RSocketRequester requester) { - // Trigger connection and sending of SETUP frame - requester.route("any-route").data("any-data").send().block(); + // Trigger connection establishment + requester.rsocketClient().source().block(); return new DefaultConnectionSetupPayload(this.connection.setupFrame()); } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java index dc9b116ac7..b2d1bf2603 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java @@ -94,7 +94,7 @@ class RSocketBufferLeakTests { @AfterAll void tearDownOnce() { - requester.dispose(); + requester.rsocketClient().dispose(); server.dispose(); context.close(); } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java index ddfb7a61f1..e6bb2cb2a1 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java @@ -96,7 +96,7 @@ public class RSocketClientToServerIntegrationTests { @AfterAll public static void tearDownOnce() { - requester.dispose(); + requester.rsocketClient().dispose(); server.dispose(); } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java index 9269af6242..29968cafd0 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java @@ -110,14 +110,14 @@ public class RSocketServerToClientIntegrationTests { .rsocketConnector(connector -> connector.acceptor(responder)) .tcp("localhost", server.address().getPort()); - // Make a request to cause a connection to be established. - requester.route("fnf").send().block(); + // Trigger connection establishment. + requester.rsocketClient().source().block(); context.getBean(ServerController.class).await(Duration.ofSeconds(5)); } finally { if (requester != null) { - requester.dispose(); + requester.rsocketClient().dispose(); } } } diff --git a/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketClientToServerCoroutinesIntegrationTests.kt b/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketClientToServerCoroutinesIntegrationTests.kt index 83cae67295..af2dc9f9e5 100644 --- a/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketClientToServerCoroutinesIntegrationTests.kt +++ b/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketClientToServerCoroutinesIntegrationTests.kt @@ -271,7 +271,7 @@ class RSocketClientToServerCoroutinesIntegrationTests { @AfterAll @JvmStatic fun tearDownOnce() { - requester.dispose() + requester.rsocketClient().dispose() server.dispose() } }