Add RSocketGraphQlClient and GraphQlRSocketHandler

See gh-339

build.gradle

@@ -63,6 +63,7 @@ configure(moduleProjects) {
 			mavenBom "org.springframework.data:spring-data-bom:2021.2.0-M4"
 			mavenBom "org.springframework.security:spring-security-bom:5.7.0-M3"
 			mavenBom "com.querydsl:querydsl-bom:5.0.0"
+			mavenBom "io.rsocket:rsocket-bom:1.1.1"
 			mavenBom "org.jetbrains.kotlin:kotlin-bom:1.5.32"
 			mavenBom "org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.5.2"
 			mavenBom "org.junit:junit-bom:5.8.1"

spring-graphql/build.gradle

@@ -11,6 +11,7 @@ dependencies {
 	compileOnly 'org.springframework:spring-webflux'
 	compileOnly 'org.springframework:spring-webmvc'
 	compileOnly 'org.springframework:spring-websocket'
+	compileOnly 'org.springframework:spring-messaging'
 	compileOnly 'javax.servlet:javax.servlet-api'
 	compileOnly 'javax.validation:validation-api'
 
@@ -19,6 +20,9 @@ dependencies {
 	compileOnly 'com.querydsl:querydsl-core'
 	compileOnly 'org.springframework.data:spring-data-commons'
 
+	compileOnly 'io.rsocket:rsocket-core'
+	compileOnly 'io.rsocket:rsocket-transport-netty'
+
 	compileOnly 'com.google.code.findbugs:jsr305'
 	compileOnly 'org.jetbrains.kotlin:kotlin-stdlib'
 	compileOnly 'org.jetbrains.kotlinx:kotlinx-coroutines-core'
@@ -29,10 +33,11 @@ dependencies {
 	testImplementation 'org.assertj:assertj-core'
 	testImplementation 'org.mockito:mockito-core'
 	testImplementation 'io.projectreactor:reactor-test'
+	testImplementation 'org.springframework:spring-messaging'
+	testImplementation 'org.springframework:spring-test'
 	testImplementation 'org.springframework:spring-webflux'
 	testImplementation 'org.springframework:spring-webmvc'
 	testImplementation 'org.springframework:spring-websocket'
-	testImplementation 'org.springframework:spring-test'
 	testImplementation 'org.springframework.data:spring-data-commons'
 	testImplementation 'org.springframework.data:spring-data-keyvalue'
 	testImplementation 'org.springframework.data:spring-data-jpa'
@@ -49,6 +54,7 @@ dependencies {
 	testImplementation 'com.querydsl:querydsl-collections'
 	testImplementation 'javax.servlet:javax.servlet-api'
 	testImplementation 'com.squareup.okhttp3:mockwebserver:3.14.9'
+	testImplementation 'io.rsocket:rsocket-transport-local'
 	testImplementation 'javax.validation:validation-api'
 	testImplementation 'com.jayway.jsonpath:json-path'
 	testImplementation 'com.fasterxml.jackson.core:jackson-databind'

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

@@ -50,7 +50,7 @@ import org.springframework.util.ClassUtils;
  */
 public abstract class AbstractGraphQlClientBuilder<B extends AbstractGraphQlClientBuilder<B>> implements GraphQlClient.Builder<B> {
 
-	private static final boolean jackson2Present = ClassUtils.isPresent(
+	protected static final boolean jackson2Present = ClassUtils.isPresent(
 			"com.fasterxml.jackson.databind.ObjectMapper", AbstractGraphQlClientBuilder.class.getClassLoader());
 
 
@@ -111,6 +111,20 @@ public abstract class AbstractGraphQlClientBuilder<B extends AbstractGraphQlClie
 		this.jsonDecoder = decoder;
 	}
 
+	/**
+	 * Variant of {@link #setJsonCodecs} for setting each codec individually.
+	 */
+	protected void setJsonEncoder(Encoder<?> encoder) {
+		this.jsonEncoder = encoder;
+	}
+
+	/**
+	 * Variant of {@link #setJsonCodecs} for setting each codec individually.
+	 */
+	protected void setJsonDecoder(Decoder<?> decoder) {
+		this.jsonDecoder = decoder;
+	}
+
 	/**
 	 * Return the configured interceptors. For subclasses that look for a
 	 * transport specific interceptor extensions.
@@ -126,8 +140,8 @@ public abstract class AbstractGraphQlClientBuilder<B extends AbstractGraphQlClie
 	protected GraphQlClient buildGraphQlClient(GraphQlTransport transport) {
 
 		if (jackson2Present) {
-			this.jsonEncoder = (this.jsonEncoder == null ? Jackson2Configurer.encoder() : this.jsonEncoder);
-			this.jsonDecoder = (this.jsonDecoder == null ? Jackson2Configurer.decoder() : this.jsonDecoder);
+			this.jsonEncoder = (this.jsonEncoder == null ? DefaultJackson2Codecs.encoder() : this.jsonEncoder);
+			this.jsonDecoder = (this.jsonDecoder == null ? DefaultJackson2Codecs.decoder() : this.jsonDecoder);
 		}
 
 		return new DefaultGraphQlClient(
@@ -178,7 +192,7 @@ public abstract class AbstractGraphQlClientBuilder<B extends AbstractGraphQlClie
 	}
 
 
-	private static class Jackson2Configurer {
+	protected static class DefaultJackson2Codecs {
 
 		static Encoder<?> encoder() {
 			return new Jackson2JsonEncoder();

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

@@ -15,6 +15,9 @@
  */
 package org.springframework.graphql.client;
 
+import java.util.List;
+import java.util.stream.Stream;
+
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.Encoder;
@@ -32,8 +35,7 @@ import org.springframework.web.reactive.socket.WebSocketMessage;
 import org.springframework.web.reactive.socket.WebSocketSession;
 
 /**
- * Delegate that can be embedded in a class to help with encoding and decoding
- * GraphQL over WebSocket messages.
+ * Helper class for encoding and decoding GraphQL messages.
  *
  * @author Rossen Stoyanchev
  * @since 1.0.0
@@ -61,22 +63,40 @@ final class CodecDelegate {
 		this.encoder = findJsonEncoder(configurer);
 	}
 
-	static Decoder<?> findJsonDecoder(CodecConfigurer configurer) {
-		return configurer.getReaders().stream()
-				.filter((reader) -> reader.canRead(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
-				.map((reader) -> ((DecoderHttpMessageReader<?>) reader).getDecoder())
-				.findFirst()
-				.orElseThrow(() -> new IllegalArgumentException("No JSON Decoder"));
+	static Encoder<?> findJsonEncoder(CodecConfigurer configurer) {
+		return findJsonEncoder(configurer.getWriters().stream()
+				.filter(writer -> writer instanceof EncoderHttpMessageWriter)
+				.map(writer -> ((EncoderHttpMessageWriter<?>) writer).getEncoder()));
 	}
 
-	static Encoder<?> findJsonEncoder(CodecConfigurer configurer) {
-		return configurer.getWriters().stream()
-				.filter((writer) -> writer.canWrite(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
-				.map((writer) -> ((EncoderHttpMessageWriter<?>) writer).getEncoder())
+	static Decoder<?> findJsonDecoder(CodecConfigurer configurer) {
+		return findJsonDecoder(configurer.getReaders().stream()
+				.filter(reader -> reader instanceof DecoderHttpMessageReader)
+				.map(reader -> ((DecoderHttpMessageReader<?>) reader).getDecoder()));
+	}
+
+	static Encoder<?> findJsonEncoder(List<Encoder<?>> encoders) {
+		return findJsonEncoder(encoders.stream());
+	}
+
+	static Decoder<?> findJsonDecoder(List<Decoder<?>> decoders) {
+		return findJsonDecoder(decoders.stream());
+	}
+
+	private static Encoder<?> findJsonEncoder(Stream<Encoder<?>> stream) {
+		return stream
+				.filter(encoder -> encoder.canEncode(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
 				.findFirst()
 				.orElseThrow(() -> new IllegalArgumentException("No JSON Encoder"));
 	}
 
+	private static Decoder<?> findJsonDecoder(Stream<Decoder<?>> decoderStream) {
+		return decoderStream
+				.filter(decoder -> decoder.canDecode(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
+				.findFirst()
+				.orElseThrow(() -> new IllegalArgumentException("No JSON Decoder"));
+	}
+
 
 	public CodecConfigurer getCodecConfigurer() {
 		return this.codecConfigurer;

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

@@ -130,6 +130,7 @@ final class DefaultHttpGraphQlClient extends AbstractDelegatingGraphQlClient imp
 		@Override
 		public HttpGraphQlClient build() {
 
+			// Pass the codecs to the parent for response decoding
 			this.webClientBuilder.codecs(configurer ->
 					setJsonCodecs(
 							CodecDelegate.findJsonEncoder(configurer),

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

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2002-2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.client;
+
+import java.net.URI;
+import java.util.function.Consumer;
+
+import io.rsocket.transport.ClientTransport;
+import io.rsocket.transport.netty.client.TcpClientTransport;
+import io.rsocket.transport.netty.client.WebsocketClientTransport;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+
+
+/**
+ * Default {@link RSocketGraphQlClient} implementation that builds the underlying
+ * {@code RSocketGraphQlTransport} to use.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+final class DefaultRSocketGraphQlClient extends AbstractDelegatingGraphQlClient implements RSocketGraphQlClient {
+
+	private final RSocketRequester.Builder requesterBuilder;
+
+	private final ClientTransport clientTransport;
+
+	private final String route;
+
+	private final Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer;
+
+
+	DefaultRSocketGraphQlClient(
+			GraphQlClient graphQlClient, RSocketRequester.Builder requesterBuilder,
+			ClientTransport clientTransport, String route,
+			Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer) {
+
+		super(graphQlClient);
+
+		this.requesterBuilder = requesterBuilder;
+		this.clientTransport = clientTransport;
+		this.route = route;
+		this.builderInitializer = builderInitializer;
+	}
+
+
+	@Override
+	public RSocketGraphQlClient.Builder<?> mutate() {
+		Builder builder = new Builder(this.requesterBuilder);
+		builder.clientTransport(this.clientTransport);
+		builder.route(this.route);
+		this.builderInitializer.accept(builder);
+		return builder;
+	}
+
+
+	/**
+	 * Default {@link RSocketGraphQlClient.Builder} implementation.
+	 */
+	static final class Builder extends AbstractGraphQlClientBuilder<Builder> implements RSocketGraphQlClient.Builder<Builder> {
+
+		private final RSocketRequester.Builder requesterBuilder;
+
+		@Nullable
+		private ClientTransport clientTransport;
+
+		private String route;
+
+		Builder() {
+			this(initRSocketRequestBuilder());
+		}
+
+		Builder(RSocketRequester.Builder requesterBuilder) {
+			Assert.notNull(requesterBuilder, "RSocketRequester.Builder is required");
+			this.requesterBuilder = requesterBuilder;
+			this.route = "graphql";
+		}
+
+		private static RSocketRequester.Builder initRSocketRequestBuilder() {
+			MimeType mimeType = MimeType.valueOf("application/graphql+json");
+			RSocketRequester.Builder requesterBuilder = RSocketRequester.builder().dataMimeType(mimeType);
+			if (jackson2Present) {
+				requesterBuilder.rsocketStrategies(
+						RSocketStrategies.builder()
+								.encoder(DefaultJackson2Codecs.encoder())
+								.decoder(DefaultJackson2Codecs.decoder())
+								.build());
+			}
+			return requesterBuilder;
+		}
+
+		@Override
+		public Builder tcp(String host, int port) {
+			this.clientTransport = TcpClientTransport.create(host, port);
+			return this;
+		}
+
+		@Override
+		public Builder webSocket(URI uri) {
+			this.clientTransport = WebsocketClientTransport.create(uri);
+			return this;
+		}
+
+		@Override
+		public Builder clientTransport(ClientTransport clientTransport) {
+			this.clientTransport = clientTransport;
+			return this;
+		}
+
+		@Override
+		public Builder dataMimeType(MimeType dataMimeType) {
+			this.requesterBuilder.dataMimeType(dataMimeType);
+			return this;
+		}
+
+		@Override
+		public Builder route(String route) {
+			Assert.notNull(route, "'route' is required");
+			this.route = route;
+			return this;
+		}
+
+		@Override
+		public Builder rsocketRequester(Consumer<RSocketRequester.Builder> requesterConsumer) {
+			requesterConsumer.accept(this.requesterBuilder);
+			return this;
+		}
+
+		@Override
+		public RSocketGraphQlClient build() {
+
+			Assert.state(this.clientTransport != null, "Neither WebSocket nor TCP networking configured");
+			RSocketRequester requester = this.requesterBuilder.transport(this.clientTransport);
+			RSocketGraphQlTransport graphQlTransport = new RSocketGraphQlTransport(this.route, requester);
+
+			// Pass the codecs to the parent for response decoding
+			this.requesterBuilder.rsocketStrategies(builder -> {
+				builder.decoders(decoders -> setJsonDecoder(CodecDelegate.findJsonDecoder(decoders)));
+				builder.encoders(encoders -> setJsonEncoder(CodecDelegate.findJsonEncoder(encoders)));
+			});
+
+			return new DefaultRSocketGraphQlClient(
+					super.buildGraphQlClient(graphQlTransport),
+					this.requesterBuilder, this.clientTransport, this.route, getBuilderInitializer());
+		}
+	}
+
+}

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

@@ -31,8 +31,9 @@ import org.springframework.web.reactive.function.client.WebClient;
 
 /**
  * Transport to execute GraphQL requests over HTTP via {@link WebClient}.
- * Supports only single-response requests over HTTP POST. For subscription
- * requests, see {@link WebSocketGraphQlTransport}.
+ *
+ * <p>Supports only single-response requests over HTTP POST. For subscriptions,
+ * see {@link WebSocketGraphQlTransport} and {@link RSocketGraphQlTransport}.
  *
  * @author Rossen Stoyanchev
  * @since 1.0.0

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

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2002-2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.client;
+
+import java.net.URI;
+import java.util.function.Consumer;
+
+import io.rsocket.transport.ClientTransport;
+
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.util.MimeType;
+
+
+/**
+ * GraphQL over RSocket client that uses {@link RSocketRequester}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+public interface RSocketGraphQlClient extends GraphQlClient {
+
+
+	@Override
+	Builder<?> mutate();
+
+
+	/**
+	 * Start with a new {@link RSocketRequester.Builder} customized for GraphQL,
+	 * setting the {@code dataMimeType} to {@code "application/graphql+json"}
+	 * and adding JSON codecs.
+	 */
+	static Builder<?> builder() {
+		return new DefaultRSocketGraphQlClient.Builder();
+	}
+
+	/**
+	 * Start with a given {@link #builder()}.
+	 */
+	static Builder<?> builder(RSocketRequester.Builder requesterBuilder) {
+		return new DefaultRSocketGraphQlClient.Builder(requesterBuilder);
+	}
+
+
+	/**
+	 * Builder for the GraphQL over HTTP client.
+	 */
+	interface Builder<B extends Builder<B>> extends GraphQlClient.Builder<B> {
+
+		/**
+		 * Select TCP as the underlying network protocol.
+		 * @param host the remote host to connect to
+		 * @param port the remote port to connect to
+		 * @return the same builder instance
+		 */
+		B tcp(String host, int port);
+
+		/**
+		 * Select WebSocket as the underlying network protocol.
+		 * @param uri the URL for the WebSocket handshake
+		 * @return the same builder instance
+		 */
+		B webSocket(URI uri);
+
+		/**
+		 * Use a given {@link ClientTransport} to communicate with the remote server.
+		 * @param clientTransport the transport to use
+		 * @return the same builder instance
+		 */
+		B clientTransport(ClientTransport clientTransport);
+
+		/**
+		 * Customize the format of data payloads for the connection.
+		 * <p>By default, this is set to {@code "application/graphql+json"} but
+		 * it can be changed to {@code "application/json"} if necessary.
+		 * @param dataMimeType the mime type to use
+		 * @return the same builder instance
+		 */
+		B dataMimeType(MimeType dataMimeType);
+
+		/**
+		 * Customize the route to specify in the metadata of each request so the
+		 * server can route it to the handler for GraphQL requests.
+		 * @param route the route
+		 * @return the same builder instance
+		 */
+		B route(String route);
+
+		/**
+		 * Customize the underlying {@code RSocketRequester} to use.
+		 * <p>Note that some properties of {@code RSocketRequester.Builder} like the
+		 * data MimeType, and the underlying RSocket transport can be customized
+		 * through this builder.
+		 * @see #dataMimeType(MimeType)
+		 * @see #tcp(String, int)
+		 * @see #webSocket(URI)
+		 * @see #clientTransport(ClientTransport)
+		 * @return the same builder instance
+		 */
+		B rsocketRequester(Consumer<RSocketRequester.Builder> requester);
+
+		/**
+		 * Build the {@code RSocketGraphQlClient} instance.
+		 */
+		@Override
+		RSocketGraphQlClient build();
+
+	}
+
+}

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

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2002-2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.client;
+
+import java.util.Map;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.graphql.GraphQlRequest;
+import org.springframework.graphql.GraphQlResponse;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.util.Assert;
+
+
+/**
+ * Transport to execute GraphQL requests over RSocket via {@link RSocketRequester}.
+ *
+ * <p>Servers are expected to support the
+ * <a href="https://github.com/rsocket/rsocket/blob/master/Extensions/Routing.md">Routing</a>
+ * metadata extension.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+final class RSocketGraphQlTransport implements GraphQlTransport {
+
+	private static final ParameterizedTypeReference<Map<String, Object>> MAP_TYPE =
+			new ParameterizedTypeReference<Map<String, Object>>() {};
+
+
+	private final String route;
+
+	private final RSocketRequester rsocketRequester;
+
+
+	RSocketGraphQlTransport(String route, RSocketRequester requester) {
+		Assert.notNull(route, "'route' is required");
+		Assert.notNull(requester, "RSocketRequester is required");
+		this.route = route;
+		this.rsocketRequester = requester;
+	}
+
+
+	@Override
+	public Mono<GraphQlResponse> execute(GraphQlRequest request) {
+		return this.rsocketRequester.route(this.route).data(request.toMap())
+				.retrieveMono(MAP_TYPE)
+				.map(ResponseMapGraphQlResponse::new);
+	}
+
+	@Override
+	public Flux<GraphQlResponse> executeSubscription(GraphQlRequest request) {
+		return this.rsocketRequester.route(this.route).data(request.toMap())
+				.retrieveFlux(MAP_TYPE)
+				.map(ResponseMapGraphQlResponse::new);
+	}
+
+}

spring-graphql/src/main/java/org/springframework/graphql/web/GraphQlRSocketHandler.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2020-2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.web;
+
+
+import java.util.Map;
+
+import graphql.ExecutionResult;
+import io.rsocket.exceptions.RejectedException;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.graphql.ExecutionGraphQlRequest;
+import org.springframework.graphql.ExecutionGraphQlResponse;
+import org.springframework.graphql.ExecutionGraphQlService;
+import org.springframework.graphql.support.DefaultExecutionGraphQlRequest;
+
+
+/**
+ * Handler for GraphQL over RSocket requests.
+ *
+ * <p>This class can be extended from an {@code @Controller} that overrides
+ * {@link #handle(Map)} and {@link #handleSubscription(Map)} in order to add
+ * {@link org.springframework.messaging.handler.annotation.MessageMapping @MessageMapping}
+ * annotations with the route.
+ *
+ * <pre style="class">
+ * &#064;Controller
+ * private static class GraphQlRSocketController extends GraphQlRSocketHandler {
+ *
+ *    GraphQlRSocketController(ExecutionGraphQlService graphQlService) {
+ *        super(graphQlService);
+ *    }
+ *
+ *    &#064;Override
+ *    &#064;MessageMapping("graphql")
+ *    public Mono<Map<String, Object>> handle(Map<String, Object> payload) {
+ *        return super.handle(payload);
+ *    }
+ *
+ *    &#064;Override
+ *    &#064;MessageMapping("graphql")
+ *    public Flux<Map<String, Object>> handleSubscription(Map<String, Object> payload) {
+ *        return super.handleSubscription(payload);
+ *    }
+ * }
+ * </pre>
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+public class GraphQlRSocketHandler {
+
+	private final ExecutionGraphQlService service;
+
+
+	public GraphQlRSocketHandler(ExecutionGraphQlService service) {
+		this.service = service;
+	}
+
+
+	/**
+	 * Handle a {@code Request-Response} interaction. For queries and mutations.
+	 */
+	public Mono<Map<String, Object>> handle(Map<String, Object> payload) {
+		return this.service.execute(initRequest(payload)).map(ExecutionGraphQlResponse::toMap);
+	}
+
+	/**
+	 * Handle a {@code Request-Stream} interaction. For subscriptions.
+	 */
+	public Flux<Map<String, Object>> handleSubscription(Map<String, Object> payload) {
+		return this.service.execute(initRequest(payload))
+				.flatMapMany(response -> {
+					if (response.getData() instanceof Publisher) {
+						Publisher<ExecutionResult> publisher = response.getData();
+						return Flux.from(publisher).map(ExecutionResult::toSpecification);
+					}
+
+					String message = (!response.isValid() ?
+							response.toMap().get("errors").toString() :
+							"Response is not a stream, is the operation actually a subscription?");
+
+					return Flux.error(new RejectedException(message));
+				});
+	}
+
+	@SuppressWarnings("unchecked")
+	private ExecutionGraphQlRequest initRequest(Map<String, Object> payload) {
+		String query = (String) payload.get("query");
+		String operationName = (String) payload.get("operationName");
+		Map<String, Object> variables = (Map<String, Object>) payload.get("variables");
+		return new DefaultExecutionGraphQlRequest(query, operationName, variables, "1", null);
+	}
+
+}

spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientBuilderTests.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2002-2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.client;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import graphql.ExecutionInput;
+import graphql.ExecutionResult;
+import graphql.ExecutionResultImpl;
+import io.rsocket.SocketAcceptor;
+import io.rsocket.core.RSocketServer;
+import io.rsocket.transport.local.LocalClientTransport;
+import io.rsocket.transport.local.LocalServerTransport;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.graphql.ExecutionGraphQlResponse;
+import org.springframework.graphql.ExecutionGraphQlService;
+import org.springframework.graphql.GraphQlRequest;
+import org.springframework.graphql.support.DefaultExecutionGraphQlResponse;
+import org.springframework.graphql.web.GraphQlRSocketHandler;
+import org.springframework.http.codec.json.Jackson2JsonDecoder;
+import org.springframework.http.codec.json.Jackson2JsonEncoder;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
+import org.springframework.stereotype.Controller;
+import org.springframework.util.Assert;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Rossen Stoyanchev
+ */
+public class RSocketGraphQlClientBuilderTests {
+
+	private static final String DOCUMENT = "{ Query }";
+
+	private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+
+	private final BuilderSetup builderSetup = new BuilderSetup();
+
+
+	@Test
+	void mutate() {
+
+		// Original
+		RSocketGraphQlClient.Builder<?> builder = this.builderSetup.initBuilder();
+		RSocketGraphQlClient client = builder.build();
+		client.document(DOCUMENT).execute().block(TIMEOUT);
+
+		GraphQlRequest request = this.builderSetup.getGraphQlRequest();
+		assertThat(request).isNotNull();
+
+		// Mutate
+		client = client.mutate().build();
+		client.document(DOCUMENT).execute().block(TIMEOUT);
+
+		request = this.builderSetup.getGraphQlRequest();
+		assertThat(request).isNotNull();
+	}
+
+
+
+	private static class BuilderSetup  {
+
+		private GraphQlRequest graphQlRequest;
+
+		private final Map<String, ExecutionGraphQlResponse> responses = new HashMap<>();
+
+		public BuilderSetup() {
+
+			ExecutionGraphQlResponse defaultResponse = new DefaultExecutionGraphQlResponse(
+					ExecutionInput.newExecutionInput().query(DOCUMENT).build(),
+					ExecutionResultImpl.newExecutionResult().build());
+
+			this.responses.put(DOCUMENT, defaultResponse);
+		}
+
+		public RSocketGraphQlClient.Builder<?> initBuilder() {
+
+			ExecutionGraphQlService graphQlService = request -> {
+				this.graphQlRequest = request;
+				String document = request.getDocument();
+				ExecutionGraphQlResponse response = this.responses.get(document);
+				Assert.notNull(response, "Unexpected request: " + document);
+				return Mono.just(response);
+			};
+
+			GraphQlRSocketController controller = new GraphQlRSocketController(graphQlService);
+
+			RSocketServer.create()
+					.acceptor(createSocketAcceptor(controller))
+					.bind(LocalServerTransport.create("local"))
+					.block();
+
+			return RSocketGraphQlClient.builder()
+					.clientTransport(LocalClientTransport.create("local"));
+		}
+
+		private SocketAcceptor createSocketAcceptor(GraphQlRSocketController controller) {
+
+			RSocketStrategies.Builder builder = RSocketStrategies.builder();
+			builder.encoder(new Jackson2JsonEncoder());
+			builder.decoder(new Jackson2JsonDecoder());
+
+			RSocketMessageHandler handler = new RSocketMessageHandler();
+			handler.setHandlers(Collections.singletonList(controller));
+			handler.setRSocketStrategies(builder.build());
+			handler.afterPropertiesSet();
+
+			return handler.responder();
+		}
+
+		public void setMockResponse(String document, ExecutionResult result) {
+			ExecutionInput executionInput = ExecutionInput.newExecutionInput().query(document).build();
+			this.responses.put(document, new DefaultExecutionGraphQlResponse(executionInput, result));
+		}
+
+		public GraphQlRequest getGraphQlRequest() {
+			return this.graphQlRequest;
+		}
+	}
+
+
+	@Controller
+	private static class GraphQlRSocketController extends GraphQlRSocketHandler {
+
+		GraphQlRSocketController(ExecutionGraphQlService service) {
+			super(service);
+		}
+
+		@Override
+		@MessageMapping("graphql")
+		public Mono<Map<String, Object>> handle(Map<String, Object> payload) {
+			return super.handle(payload);
+		}
+
+		@Override
+		@MessageMapping("graphql")
+		public Flux<Map<String, Object>> handleSubscription(Map<String, Object> payload) {
+			return super.handleSubscription(payload);
+		}
+	}
+
+}

spring-graphql/src/test/resources/log4j2-test.xml

@@ -9,6 +9,7 @@
 		<Logger name="org.springframework" level="debug" />
 		<Logger name="org.springframework.graphql" level="trace" />
 		<Logger name="graphql" level="info" />
+		<Logger name="io.rsocket" level="info" />
 		<Root level="error">
 			<AppenderRef ref="Console" />
 		</Root>