Add start and stop for RSocket client and tester

Closes gh-378

Author: rstoyanchev

spring-graphql-docs/src/docs/asciidoc/client.adoc

@@ -178,16 +178,17 @@ to execute GraphQL requests over RSocket requests.
 ----
 
 In contrast to `HttpGraphQlClient`, the `RSocketGraphQlClient` is connection oriented,
-which means it needs to establish a connection before making any requests. As you begin
-to make requests, the connection is established transparently.
+which means it needs to establish a session before making any requests. As you begin
+to make requests, the session is established transparently. Alternatively, use the
+client's `start()` method to establish the session explicitly before any requests.
 
-`RSocketGraphQlClient` is also multiplexed. It maintains a single, shared connection for
-all requests. If the connection is lost, it is re-established on the next request. You
-can use the `dispose()` method on the underlying `RSocketRequester` to close the
-connection explicitly.
+`RSocketGraphQlClient` is also multiplexed. It maintains a single, shared session for
+all requests.  If the session is lost, it is re-established on the next request or if
+`start()` is called again. You can also use the client's `stop()` method which cancels
+in-progress requests, closes the session, and rejects new requests.
 
 TIP: Use a single `RSocketGraphQlClient` instance for each server in order to have a
-single, shared connection for all requests to that server. Each client instance
+single, shared session for all requests to that server. Each client instance
 establishes its own connection and that is typically not the intent for a single server.
 
 Once `RSocketGraphQlClient` is created, you can begin to

spring-graphql-docs/src/docs/asciidoc/testing.adoc

@@ -202,9 +202,9 @@ requests over RSocket:
 ----
 
 `RSocketGraphQlTester` is connection oriented and multiplexed. Each instance establishes
-its own single, shared connection for all requests. Typically, you'll want to use a single
-instance only per server. You can use the `dispose()` method on the underlying
-`RSocketRequester` to close the connection explicitly.
+its own single, shared session for all requests. Typically, you'll want to use a single
+instance only per server. You can use the `stop()` method on the tester to close the
+session explicitly.
 
 Once `RSocketGraphQlTester` is created, you can begin to
 <<testing-requests, execute requests>> using the same API, independent of the underlying

spring-graphql-test/src/main/java/org/springframework/graphql/test/tester/DefaultRSocketGraphQlTesterBuilder.java

@@ -21,6 +21,7 @@
 import java.util.function.Consumer;
 
 import io.rsocket.transport.ClientTransport;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.Encoder;
@@ -141,6 +142,16 @@ private static class DefaultRSocketGraphQlTester extends AbstractDelegatingGraph
 			this.builderInitializer = builderInitializer;
 		}
 
+		@Override
+		public Mono<Void> start() {
+			return this.rsocketGraphQlClient.start();
+		}
+
+		@Override
+		public Mono<Void> stop() {
+			return this.rsocketGraphQlClient.stop();
+		}
+
 		@Override
 		public RSocketGraphQlTester.Builder<?> mutate() {
 			DefaultRSocketGraphQlTesterBuilder builder = new DefaultRSocketGraphQlTesterBuilder(this.rsocketGraphQlClient);

spring-graphql-test/src/main/java/org/springframework/graphql/test/tester/RSocketGraphQlTester.java

@@ -21,7 +21,9 @@
 import java.util.function.Consumer;
 
 import io.rsocket.transport.ClientTransport;
+import reactor.core.publisher.Mono;
 
+import org.springframework.graphql.client.RSocketGraphQlClient;
 import org.springframework.messaging.rsocket.RSocketRequester;
 import org.springframework.util.MimeType;
 
@@ -33,6 +35,19 @@
  */
 public interface RSocketGraphQlTester extends GraphQlTester {
 
+	/**
+	 * Start the RSocket session.
+	 * @return {@code Mono} that completes when the underlying session is started
+	 */
+	Mono<Void> start();
+
+	/**
+	 * Stop the RSocket session.
+	 * @return {@code Mono} that completes when the underlying session is closed
+	 * <p>Note that currently this method not differed and does not wait,
+	 * see {@link RSocketGraphQlClient#stop()}
+	 */
+	Mono<Void> stop();
 
 	@Override
 	RSocketGraphQlTester.Builder<?> mutate();

spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClientBuilder.java

@@ -22,6 +22,7 @@
 import io.rsocket.transport.ClientTransport;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.rsocket.transport.netty.client.WebsocketClientTransport;
+import reactor.core.publisher.Mono;
 
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.rsocket.RSocketRequester;
@@ -124,7 +125,7 @@ public RSocketGraphQlClient build() {
 		RSocketGraphQlTransport graphQlTransport = new RSocketGraphQlTransport(this.route, requester, getJsonDecoder());
 
 		return new DefaultRSocketGraphQlClient(
-				super.buildGraphQlClient(graphQlTransport),
+				super.buildGraphQlClient(graphQlTransport), requester,
 				this.requesterBuilder, this.clientTransport, this.route, getBuilderInitializer());
 	}
 
@@ -134,6 +135,8 @@ public RSocketGraphQlClient build() {
 	 */
 	private static class DefaultRSocketGraphQlClient extends AbstractDelegatingGraphQlClient implements RSocketGraphQlClient {
 
+		private final RSocketRequester requester;
+
 		private final RSocketRequester.Builder requesterBuilder;
 
 		private final ClientTransport clientTransport;
@@ -143,17 +146,30 @@ private static class DefaultRSocketGraphQlClient extends AbstractDelegatingGraph
 		private final Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer;
 
 		DefaultRSocketGraphQlClient(
-				GraphQlClient graphQlClient, RSocketRequester.Builder requesterBuilder,
+				GraphQlClient graphQlClient, RSocketRequester requester, RSocketRequester.Builder requesterBuilder,
 				ClientTransport clientTransport, String route, Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer) {
 
 			super(graphQlClient);
 
+			this.requester = requester;
 			this.requesterBuilder = requesterBuilder;
 			this.clientTransport = clientTransport;
 			this.route = route;
 			this.builderInitializer = builderInitializer;
 		}
 
+		@Override
+		public Mono<Void> start() {
+			return this.requester.rsocketClient().source().then();
+		}
+
+		@Override
+		public Mono<Void> stop() {
+			// Currently, no option to close and wait (see Javadoc)
+			this.requester.dispose();
+			return Mono.empty();
+		}
+
 		@Override
 		public RSocketGraphQlClient.Builder<?> mutate() {
 			DefaultRSocketGraphQlClientBuilder builder = new DefaultRSocketGraphQlClientBuilder(this.requesterBuilder);

spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlClient.java

@@ -19,7 +19,9 @@
 import java.net.URI;
 import java.util.function.Consumer;
 
+import io.rsocket.core.RSocketClient;
 import io.rsocket.transport.ClientTransport;
+import reactor.core.publisher.Mono;
 
 import org.springframework.messaging.rsocket.RSocketRequester;
 import org.springframework.util.MimeType;
@@ -33,6 +35,21 @@
  */
 public interface RSocketGraphQlClient extends GraphQlClient {
 
+	/**
+	 * Start the RSocket session.
+	 * @return {@code Mono} that completes when the underlying session is started
+	 */
+	Mono<Void> start();
+
+	/**
+	 * Stop the RSocket session.
+	 * @return {@code Mono} that completes when the underlying session is stopped.
+	 * <p>Note that currently this method calls {@link RSocketClient#dispose()}
+	 * which is not differed and does not wait, i.e. it triggers stopping
+	 * immediately and returns immediately.
+	 * See <a href="https://github.com/rsocket/rsocket-java/issues/1048">rsocket-java#1048</a>
+	 */
+	Mono<Void> stop();
 
 	@Override
 	Builder<?> mutate();