Support for making requests via RSocketClient

Closes gh-25332

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java

@@ -22,6 +22,7 @@ import java.util.function.Consumer;
 
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
+import io.rsocket.RSocketClient;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -48,8 +49,7 @@ final class DefaultRSocketRequester implements RSocketRequester {
 
 	private static final Map<String, Object> EMPTY_HINTS = Collections.emptyMap();
 
-
-	private final RSocket rsocket;
+	private final RSocketDelegate rsocketDelegate;
 
 	private final MimeType dataMimeType;
 
@@ -61,15 +61,15 @@ final class DefaultRSocketRequester implements RSocketRequester {
 
 
 	DefaultRSocketRequester(
-			RSocket rsocket, MimeType dataMimeType, MimeType metadataMimeType,
+			RSocketDelegate rsocketDelegate, MimeType dataMimeType, MimeType metadataMimeType,
 			RSocketStrategies strategies) {
 
-		Assert.notNull(rsocket, "RSocket is required");
+		Assert.notNull(rsocketDelegate, "RSocket or RSocketClient is required");
 		Assert.notNull(dataMimeType, "'dataMimeType' is required");
 		Assert.notNull(metadataMimeType, "'metadataMimeType' is required");
 		Assert.notNull(strategies, "RSocketStrategies is required");
 
-		this.rsocket = rsocket;
+		this.rsocketDelegate = rsocketDelegate;
 		this.dataMimeType = dataMimeType;
 		this.metadataMimeType = metadataMimeType;
 		this.strategies = strategies;
@@ -77,9 +77,11 @@ final class DefaultRSocketRequester implements RSocketRequester {
 	}
 
 
+	@Nullable
 	@Override
 	public RSocket rsocket() {
-		return this.rsocket;
+		return (this.rsocketDelegate instanceof ConnectionRSocketDelegate ?
+				((ConnectionRSocketDelegate) this.rsocketDelegate).getRSocket() : null);
 	}
 
 	@Override
@@ -102,6 +104,10 @@ 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()));
@@ -250,12 +256,12 @@ final class DefaultRSocketRequester implements RSocketRequester {
 
 		@Override
 		public Mono<Void> sendMetadata() {
-			return getPayloadMono().flatMap(rsocket::metadataPush);
+			return rsocketDelegate().metadataPush(getPayloadMono());
 		}
 
 		@Override
 		public Mono<Void> send() {
-			return getPayloadMono().flatMap(rsocket::fireAndForget);
+			return rsocketDelegate().fireAndForget(getPayloadMono());
 		}
 
 		@Override
@@ -270,7 +276,7 @@ final class DefaultRSocketRequester implements RSocketRequester {
 
 		@SuppressWarnings("unchecked")
 		private <T> Mono<T> retrieveMono(ResolvableType elementType) {
-			Mono<Payload> payloadMono = getPayloadMono().flatMap(rsocket::requestResponse);
+			Mono<Payload> payloadMono = rsocketDelegate().requestResponse(getPayloadMono());
 
 			if (isVoid(elementType)) {
 				return (Mono<T>) payloadMono.then();
@@ -295,8 +301,8 @@ final class DefaultRSocketRequester implements RSocketRequester {
 		private <T> Flux<T> retrieveFlux(ResolvableType elementType) {
 
 			Flux<Payload> payloadFlux = (this.payloadFlux != null ?
-					rsocket.requestChannel(this.payloadFlux) :
-					getPayloadMono().flatMapMany(rsocket::requestStream));
+					rsocketDelegate().requestChannel(this.payloadFlux) :
+					rsocketDelegate().requestStream(getPayloadMono()));
 
 			if (isVoid(elementType)) {
 				return payloadFlux.thenMany(Flux.empty());
@@ -307,6 +313,10 @@ final class DefaultRSocketRequester implements RSocketRequester {
 					(T) decoder.decode(dataBuffer, elementType, dataMimeType, EMPTY_HINTS));
 		}
 
+		private RSocketDelegate rsocketDelegate() {
+			return DefaultRSocketRequester.this.rsocketDelegate;
+		}
+
 		private Mono<Payload> getPayloadMono() {
 			Assert.state(this.payloadFlux == null, "No RSocket interaction with Flux request and Mono response.");
 			return this.payloadMono != null ? this.payloadMono : firstPayload(emptyBufferMono);
@@ -316,4 +326,107 @@ final class DefaultRSocketRequester implements RSocketRequester {
 			return PayloadUtils.retainDataAndReleasePayload(payload, bufferFactory());
 		}
 	}
+
+
+	// Contract to avoid a hard dependency on RSocketClient for now.
+
+	interface RSocketDelegate {
+
+		Mono<Void> fireAndForget(Mono<Payload> payloadMono);
+
+		Mono<Payload> requestResponse(Mono<Payload> payloadMono);
+
+		Flux<Payload> requestStream(Mono<Payload> payloadMono);
+
+		Flux<Payload> requestChannel(Publisher<Payload> payloadPublisher);
+
+		Mono<Void> metadataPush(Mono<Payload> 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<Void> fireAndForget(Mono<Payload> payloadMono) {
+			return payloadMono.flatMap(this.rsocket::fireAndForget);
+		}
+
+		@Override
+		public Mono<Payload> requestResponse(Mono<Payload> payloadMono) {
+			return payloadMono.flatMap(this.rsocket::requestResponse);
+		}
+
+		@Override
+		public Flux<Payload> requestStream(Mono<Payload> payloadMono) {
+			return payloadMono.flatMapMany(this.rsocket::requestStream);
+		}
+
+		@Override
+		public Flux<Payload> requestChannel(Publisher<Payload> payloadPublisher) {
+			return this.rsocket.requestChannel(payloadPublisher);
+		}
+
+		@Override
+		public Mono<Void> metadataPush(Mono<Payload> 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<Void> fireAndForget(Mono<Payload> payloadMono) {
+			return this.rsocketClient.fireAndForget(payloadMono);
+		}
+
+		@Override
+		public Mono<Payload> requestResponse(Mono<Payload> payloadMono) {
+			return this.rsocketClient.requestResponse(payloadMono);
+		}
+
+		@Override
+		public Flux<Payload> requestStream(Mono<Payload> payloadMono) {
+			return this.rsocketClient.requestStream(payloadMono);
+		}
+
+		@Override
+		public Flux<Payload> requestChannel(Publisher<Payload> payloadPublisher) {
+			return this.rsocketClient.requestChannel(payloadPublisher);
+		}
+
+		@Override
+		public Mono<Void> metadataPush(Mono<Payload> payloadMono) {
+			return this.rsocketClient.metadataPush(payloadMono);
+		}
+
+		@Override
+		public void dispose() {
+			this.rsocketClient.dispose();
+		}
+	}
+
 }

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilder.java

@@ -23,10 +23,9 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
-import java.util.function.Function;
 
 import io.rsocket.Payload;
-import io.rsocket.RSocket;
+import io.rsocket.core.RSocketConnector;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.metadata.WellKnownMimeType;
 import io.rsocket.transport.ClientTransport;
@@ -45,7 +44,6 @@ import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
@@ -59,11 +57,6 @@ import org.springframework.util.MimeTypeUtils;
  */
 final class DefaultRSocketRequesterBuilder implements RSocketRequester.Builder {
 
-	private final static boolean rsocketConnectorPresent =
-			ClassUtils.isPresent("io.rsocket.core.RSocketConnector",
-					DefaultRSocketRequesterBuilder.class.getClassLoader());
-
-
 	private static final Map<String, Object> HINTS = Collections.emptyMap();
 
 	private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
@@ -165,47 +158,70 @@ final class DefaultRSocketRequesterBuilder implements RSocketRequester.Builder {
 	}
 
 	@Override
+	public RSocketRequester tcp(String host, int port) {
+		return transport(TcpClientTransport.create(host, port));
+	}
+
+	@Override
+	public RSocketRequester websocket(URI uri) {
+		return transport(WebsocketClientTransport.create(uri));
+	}
+
+	@Override
+	public RSocketRequester transport(ClientTransport transport) {
+		RSocketStrategies strategies = getRSocketStrategies();
+		Assert.isTrue(!strategies.encoders().isEmpty(), "No encoders");
+		Assert.isTrue(!strategies.decoders().isEmpty(), "No decoders");
+
+		MimeType metaMimeType = (this.metadataMimeType != null ? this.metadataMimeType :
+				MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()));
+
+		MimeType dataMimeType = getDataMimeType(strategies);
+		Mono<Payload> setupPayload = getSetupPayload(dataMimeType, metaMimeType, strategies);
+
+		RSocketConnector connector = initConnector(
+				this.rsocketConnectorConfigurers, this.rsocketFactoryConfigurers,
+				metaMimeType, dataMimeType, setupPayload, strategies);
+
+		return new DefaultRSocketRequester(
+				new DefaultRSocketRequester.ClientRSocketDelegate(connector.toRSocketClient(transport)),
+				dataMimeType, metaMimeType, strategies);
+	}
+
+	@Override
+	@SuppressWarnings("deprecation")
 	public Mono<RSocketRequester> connectTcp(String host, int port) {
 		return connect(TcpClientTransport.create(host, port));
 	}
 
 	@Override
+	@SuppressWarnings("deprecation")
 	public Mono<RSocketRequester> connectWebSocket(URI uri) {
 		return connect(WebsocketClientTransport.create(uri));
 	}
 
 	@Override
+	@SuppressWarnings("deprecation")
 	public Mono<RSocketRequester> connect(ClientTransport transport) {
+
 		RSocketStrategies rsocketStrategies = getRSocketStrategies();
 		Assert.isTrue(!rsocketStrategies.encoders().isEmpty(), "No encoders");
 		Assert.isTrue(!rsocketStrategies.decoders().isEmpty(), "No decoders");
 
-		MimeType metaMimeType = this.metadataMimeType != null ? this.metadataMimeType :
-				MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString());
+		MimeType metaMimeType = (this.metadataMimeType != null ? this.metadataMimeType :
+				MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()));
 
 		MimeType dataMimeType = getDataMimeType(rsocketStrategies);
 		Mono<Payload> setupPayload = getSetupPayload(dataMimeType, metaMimeType, rsocketStrategies);
 
-		Function<Payload, Mono<RSocket>> connectFunction;
-		if (rsocketConnectorPresent) {
-			connectFunction = payload -> new RSocketConnectorHelper().getRSocketMono(
-					this.rsocketConnectorConfigurers, this.rsocketFactoryConfigurers,
-					metaMimeType, dataMimeType, setupPayload, rsocketStrategies, transport, payload);
-		}
-		else {
-			connectFunction = payload -> new RSocketFactoryHelper().getRSocketMono(
-					this.rsocketFactoryConfigurers, metaMimeType, dataMimeType,
-					setupPayload, rsocketStrategies, transport, payload);
-		}
+		RSocketConnector connector = initConnector(
+				this.rsocketConnectorConfigurers, this.rsocketFactoryConfigurers,
+				metaMimeType, dataMimeType, setupPayload, rsocketStrategies);
 
-		// In RSocket 1.0.2 we can pass a Mono for the setup Payload. Until then we have to
-		// resolve it and then cache the Mono<RSocket> because it may be a ReconnectMono.
-
-		return setupPayload
-				.map(connectFunction)
-				.cache()
-				.flatMap(mono -> mono.map(rsocket ->
-						new DefaultRSocketRequester(rsocket, dataMimeType, metaMimeType, rsocketStrategies)));
+		return connector.connect(transport).map(rsocket ->
+				new DefaultRSocketRequester(
+						new DefaultRSocketRequester.ConnectionRSocketDelegate(rsocket),
+						dataMimeType, metaMimeType, rsocketStrategies));
 	}
 
 	private RSocketStrategies getRSocketStrategies() {
@@ -289,60 +305,32 @@ final class DefaultRSocketRequesterBuilder implements RSocketRequester.Builder {
 				.doOnDiscard(Payload.class, Payload::release);
 	}
 
-
 	@SuppressWarnings("deprecation")
-	private static class RSocketConnectorHelper {
+	private RSocketConnector initConnector(List<RSocketConnectorConfigurer> connectorConfigurers,
+			List<ClientRSocketFactoryConfigurer> factoryConfigurers,
+			MimeType metaMimeType, MimeType dataMimeType, Mono<Payload> setupPayloadMono,
+			RSocketStrategies rsocketStrategies) {
 
-		Mono<RSocket> getRSocketMono(List<RSocketConnectorConfigurer> connectorConfigurers,
-				List<ClientRSocketFactoryConfigurer> factoryConfigurers,
-				MimeType metaMimeType, MimeType dataMimeType, Mono<Payload> setupPayload,
-				RSocketStrategies rsocketStrategies, ClientTransport transport, Payload payload) {
+		RSocketConnector connector = RSocketConnector.create();
+		connectorConfigurers.forEach(c -> c.configure(connector));
 
-			io.rsocket.core.RSocketConnector connector = io.rsocket.core.RSocketConnector.create();
-			connectorConfigurers.forEach(c -> c.configure(connector));
-
-			if (!factoryConfigurers.isEmpty()) {
-				io.rsocket.RSocketFactory.ClientRSocketFactory factory =
-						new io.rsocket.RSocketFactory.ClientRSocketFactory(connector);
-				factoryConfigurers.forEach(c -> c.configure(factory));
-			}
-
-			if (rsocketStrategies.dataBufferFactory() instanceof NettyDataBufferFactory) {
-				connector.payloadDecoder(PayloadDecoder.ZERO_COPY);
-			}
-
-			connector.metadataMimeType(metaMimeType.toString());
-			connector.dataMimeType(dataMimeType.toString());
-
-			if (setupPayload != EMPTY_SETUP_PAYLOAD) {
-				connector.setupPayload(payload);
-			}
-			return connector.connect(transport);
+		if (!factoryConfigurers.isEmpty()) {
+			io.rsocket.RSocketFactory.ClientRSocketFactory factory =
+					new io.rsocket.RSocketFactory.ClientRSocketFactory(connector);
+			factoryConfigurers.forEach(c -> c.configure(factory));
 		}
-	}
 
-
-	@SuppressWarnings("deprecation")
-	private static class RSocketFactoryHelper {
-
-		Mono<RSocket> getRSocketMono(List<ClientRSocketFactoryConfigurer> configurers,
-				MimeType metaMimeType, MimeType dataMimeType, Mono<Payload> setupPayload,
-				RSocketStrategies rsocketStrategies, ClientTransport transport, Payload payload) {
-
-			io.rsocket.RSocketFactory.ClientRSocketFactory factory = io.rsocket.RSocketFactory.connect();
-			configurers.forEach(c -> c.configure(factory));
-
-			if (rsocketStrategies.dataBufferFactory() instanceof NettyDataBufferFactory) {
-				factory.frameDecoder(PayloadDecoder.ZERO_COPY);
-			}
-
-			factory.metadataMimeType(metaMimeType.toString());
-			factory.dataMimeType(dataMimeType.toString());
-			if (setupPayload != EMPTY_SETUP_PAYLOAD) {
-				factory.setupPayload(payload);
-			}
-			return factory.transport(transport).start();
+		if (rsocketStrategies.dataBufferFactory() instanceof NettyDataBufferFactory) {
+			connector.payloadDecoder(PayloadDecoder.ZERO_COPY);
 		}
+
+		connector.metadataMimeType(metaMimeType.toString());
+		connector.dataMimeType(dataMimeType.toString());
+
+		if (setupPayloadMono != EMPTY_SETUP_PAYLOAD) {
+			connector.setupPayload(setupPayloadMono);
+		}
+		return connector;
 	}
 
 }

spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java

@@ -48,8 +48,15 @@ import org.springframework.util.MimeType;
 public interface RSocketRequester {
 
 	/**
-	 * Return the underlying sending RSocket.
+	 * 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}.
 	 */
+	@Nullable
 	RSocket rsocket();
 
 	/**
@@ -96,6 +103,12 @@ 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
@@ -113,7 +126,9 @@ public interface RSocketRequester {
 			RSocket rsocket, MimeType dataMimeType, MimeType metadataMimeType,
 			RSocketStrategies strategies) {
 
-		return new DefaultRSocketRequester(rsocket, dataMimeType, metadataMimeType, strategies);
+		return new DefaultRSocketRequester(
+				new DefaultRSocketRequester.ConnectionRSocketDelegate(rsocket),
+				dataMimeType, metadataMimeType, strategies);
 	}
 
 
@@ -236,28 +251,65 @@ public interface RSocketRequester {
 		 */
 		RSocketRequester.Builder apply(Consumer<RSocketRequester.Builder> configurer);
 
+		/**
+		 * Build an {@link RSocketRequester} instance for use with a TCP
+		 * transport. Requests are made via {@link io.rsocket.RSocketClient}
+		 * which establishes a shared TCP connection to given host and port.
+		 * @param host the host of the server to connect to
+		 * @param port the port of the server to connect to
+		 * @return the created {@code RSocketRequester}
+		 * @since 5.3
+		 */
+		RSocketRequester tcp(String host, int port);
+
+		/**
+		 * Build an {@link RSocketRequester} instance for use with a WebSocket
+		 * transport. Requests are made via {@link io.rsocket.RSocketClient}
+		 * which establishes a shared WebSocket connection to given URL.
+		 * @param uri the URL of the server to connect to
+		 * @return the created {@code RSocketRequester}
+		 * @since 5.3
+		 */
+		RSocketRequester websocket(URI uri);
+
+		/**
+		 * Build an {@link RSocketRequester} instance for use with the given
+		 * transport. Requests are made via {@link io.rsocket.RSocketClient}
+		 * which establishes a shared connection through the given transport.
+		 * @param transport the transport to use for connecting to the server
+		 * @return the created {@code RSocketRequester}
+		 * @since 5.3
+		 */
+		RSocketRequester transport(ClientTransport transport);
+
 		/**
 		 * Connect to the server over TCP.
 		 * @param host the server host
 		 * @param port the server port
 		 * @return an {@code RSocketRequester} for the connection
+		 * @deprecated as of 5.3 in favor of {@link #tcp(String, int)}
 		 * @see TcpClientTransport
 		 */
+		@Deprecated
 		Mono<RSocketRequester> connectTcp(String host, int port);
 
 		/**
 		 * Connect to the server over WebSocket.
 		 * @param uri the RSocket server endpoint URI
 		 * @return an {@code RSocketRequester} for the connection
+		 * @deprecated as of 5.3 in favor of {@link #websocket(URI)}
 		 * @see WebsocketClientTransport
 		 */
+		@Deprecated
 		Mono<RSocketRequester> connectWebSocket(URI uri);
 
 		/**
 		 * Connect to the server with the given {@code ClientTransport}.
 		 * @param transport the client transport to use
 		 * @return an {@code RSocketRequester} for the connection
+		 * @deprecated as of 5.3 in favor of {@link #transport(ClientTransport)}
 		 */
+		@Deprecated
 		Mono<RSocketRequester> connect(ClientTransport transport);
 
 	}

spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt

@@ -18,9 +18,9 @@ package org.springframework.messaging.rsocket
 
 import io.rsocket.transport.ClientTransport
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.reactive.asFlow
 import kotlinx.coroutines.reactive.awaitFirstOrNull
 import kotlinx.coroutines.reactive.awaitSingle
-import kotlinx.coroutines.reactive.asFlow
 import org.reactivestreams.Publisher
 import org.springframework.core.ParameterizedTypeReference
 import reactor.core.publisher.Flux
@@ -33,6 +33,7 @@ import java.net.URI
  * @author Sebastien Deleuze
  * @since 5.2
  */
+@Suppress("DEPRECATION")
 suspend fun RSocketRequester.Builder.connectAndAwait(transport: ClientTransport): RSocketRequester =
 		connect(transport).awaitSingle()
 
@@ -42,6 +43,7 @@ suspend fun RSocketRequester.Builder.connectAndAwait(transport: ClientTransport)
  * @author Sebastien Deleuze
  * @since 5.2
  */
+@Suppress("DEPRECATION")
 suspend fun RSocketRequester.Builder.connectTcpAndAwait(host: String, port: Int): RSocketRequester =
 		connectTcp(host, port).awaitSingle()
 
@@ -51,6 +53,7 @@ suspend fun RSocketRequester.Builder.connectTcpAndAwait(host: String, port: Int)
  * @author Sebastien Deleuze
  * @since 5.2
  */
+@Suppress("DEPRECATION")
 suspend fun RSocketRequester.Builder.connectWebSocketAndAwait(uri: URI): RSocketRequester =
 		connectWebSocket(uri).awaitSingle()