diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketClient.java b/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketClient.java
new file mode 100644
index 00000000..93fc6ab2
--- /dev/null
+++ b/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketClient.java
@@ -0,0 +1,83 @@
+/*
+ * 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.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.util.Assert;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.client.WebSocketClient;
+
+/**
+ * {@link WebSocketClient} that uses {@link TestWebSocketConnection} to connect
+ * and test the interaction between a client and a server {@link WebSocketHandler}.
+ *
+ * <p>Call {@link #execute(URI, WebSocketHandler)}, subscribe, and then use
+ * getters to access established connections by index from based on order of
+ * execution.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class TestWebSocketClient implements WebSocketClient {
+
+	private final WebSocketHandler serverHandler;
+
+	private final List<TestWebSocketConnection> connections = new CopyOnWriteArrayList<>();
+
+
+	public TestWebSocketClient(WebSocketHandler serverHandler) {
+		this.serverHandler = serverHandler;
+	}
+
+
+	/**
+	 * Return the connection at the specified index from a list of connections
+	 * based on order of execution.
+	 */
+	public TestWebSocketConnection getConnection(int index) {
+		Assert.isTrue(index < this.connections.size(),
+				"No connection at index=" + index + ", total=" + this.connections.size());
+		return connections.get(index);
+	}
+
+	/**
+	 * Return the number of connections which corresponds to the number of calls
+	 * to one of the execute methods.
+	 */
+	public int getConnectionCount() {
+		return this.connections.size();
+	}
+
+
+	@Override
+	public Mono<Void> execute(URI url, WebSocketHandler clientHandler) {
+		return execute(URI.create("/"), HttpHeaders.EMPTY, clientHandler);
+	}
+
+	@Override
+	public Mono<Void> execute(URI url, HttpHeaders headers, WebSocketHandler clientHandler) {
+		TestWebSocketConnection connection = new TestWebSocketConnection(url, headers);
+		this.connections.add(connection);
+		return connection.connect(clientHandler, this.serverHandler);
+	}
+
+}
diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketConnection.java b/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketConnection.java
new file mode 100644
index 00000000..4cd31b8b
--- /dev/null
+++ b/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketConnection.java
@@ -0,0 +1,258 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.reactivestreams.Publisher;
+import reactor.core.Scannable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.web.reactive.socket.CloseStatus;
+import org.springframework.web.reactive.socket.HandshakeInfo;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.WebSocketMessage;
+import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession;
+
+/**
+ * Emulates a WebSocket connection by connecting a client and a server handlers
+ * via two message {@link Sinks.Many sinks}, one for each end of the connection.
+ *
+ * <p>Use {@link TestWebSocketClient} to establish connections that can then be
+ * accessed through it.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class TestWebSocketConnection {
+
+	private static final AtomicLong connectionIndex = new AtomicLong();
+
+
+	private final TestWebSocketSession clientSession;
+
+	private final TestWebSocketSession serverSession;
+
+
+	public TestWebSocketConnection(URI url, HttpHeaders headers) {
+
+		long id = connectionIndex.incrementAndGet();
+
+		Sinks.Many<WebSocketMessage> clientSink = Sinks.many().unicast().onBackpressureBuffer();
+		Sinks.Many<WebSocketMessage> serverSink = Sinks.many().unicast().onBackpressureBuffer();
+
+		Sinks.One<CloseStatus> clientStatusSink = Sinks.one();
+		Sinks.One<CloseStatus> serverStatusSink = Sinks.one();
+
+		this.clientSession = new TestWebSocketSession("client-session-" + id, url, headers,
+				clientSink, serverSink.asFlux(), clientStatusSink, serverStatusSink.asMono());
+
+		this.serverSession = new TestWebSocketSession("server-session-" + id, url, headers,
+				serverSink, clientSink.asFlux(), serverStatusSink, clientStatusSink.asMono());
+	}
+
+
+	/**
+	 * Return {@code true} if both client and server sessions are open.
+	 */
+	public boolean isOpen() {
+		return (this.clientSession.isOpen() && this.serverSession.isOpen());
+	}
+
+	/**
+	 * Return messages sent from the client side.
+	 */
+	public List<WebSocketMessage> getClientMessages() {
+		return this.clientSession.getSentMessages();
+	}
+
+	/**
+	 * Return messages sent from the server side.
+	 */
+	public List<WebSocketMessage> getServerMessages() {
+		return this.serverSession.getSentMessages();
+	}
+
+
+	/**
+	 * Starts client and server session handling, and if either side errors or
+	 * completes, close its session with either {@link CloseStatus#NORMAL} or
+	 * {@link CloseStatus#PROTOCOL_ERROR} respectively.
+	 * @param clientHandler the client session handler
+	 * @param serverHandler the server session handler
+	 * @return {@code Mono} that completes when either client or server
+	 * session handling completes or when either is closed
+	 */
+	Mono<Void> connect(WebSocketHandler clientHandler, WebSocketHandler serverHandler) {
+
+		Mono<Void> serverMono = invokeHandler(serverHandler, this.serverSession);
+		Mono<Void> clientMono = invokeHandler(clientHandler, this.clientSession);
+
+		Mono<Void> serverStatusMono = this.serverSession.closeStatus().then();
+		Mono<Void> clientStatusMono = this.clientSession.closeStatus().then();
+
+		return Mono.zip(serverMono, clientMono, serverStatusMono, clientStatusMono).then();
+	}
+
+	/**
+	 * Handle the session and complete it when handling completes.
+	 */
+	private Mono<Void> invokeHandler(WebSocketHandler serverHandler, TestWebSocketSession session) {
+		return serverHandler.handle(session)
+				.then(Mono.defer(() -> session.close(CloseStatus.NORMAL)))
+				.onErrorResume(ex -> session.close(CloseStatus.PROTOCOL_ERROR).then(Mono.error(ex)));
+	}
+
+
+	/**
+	 * Close the connection from the client side.
+	 */
+	public Mono<Void> closeClientSession(CloseStatus status) {
+		return this.clientSession.close(status);
+	}
+
+	/**
+	 * Close the connection from the server side.
+	 */
+	public Mono<Void> closeServerSession(CloseStatus status) {
+		return this.serverSession.close(status);
+	}
+
+	/**
+	 * Return the {@code CloseStatus} that may have come from either side.
+	 */
+	public Mono<CloseStatus> closeStatus() {
+		return this.clientSession.closeStatus().or(this.serverSession.closeStatus());
+	}
+
+
+	/**
+	 * Test WebSocketSession that sends to a given {@link Sinks.Many sink} and
+	 * receives from a given {@code Flux}.
+	 */
+	private static class TestWebSocketSession extends AbstractWebSocketSession<Object> {
+
+		private final Sinks.Many<WebSocketMessage> sendSink;
+
+		private final Flux<WebSocketMessage> receiveFlux;
+
+		private final Sinks.One<CloseStatus> closeStatusSink;
+
+		private final Queue<WebSocketMessage> sentMessages = new ConcurrentLinkedQueue<>();
+
+
+		TestWebSocketSession(String sessionId, URI url, HttpHeaders headers,
+				Sinks.Many<WebSocketMessage> sendSink, Flux<WebSocketMessage> receiveFlux,
+				Sinks.One<CloseStatus> closeStatusSink, Mono<CloseStatus> remoteCloseStatusMono) {
+
+			super(new Object(), sessionId, new HandshakeInfo(url, headers, Mono.empty(), null),
+					DefaultDataBufferFactory.sharedInstance);
+
+			this.sendSink = sendSink;
+			this.receiveFlux = receiveFlux.cache();
+			this.closeStatusSink = closeStatusSink;
+
+			// Close this side when the remote closes
+			remoteCloseStatusMono.doOnSuccess(this::handleRemoteClosure).subscribe();
+		}
+
+		private void handleRemoteClosure(@Nullable CloseStatus status) {
+
+			if (!isOpen()) {
+				// when we close, remote closes, and we detect that
+				return;
+			}
+
+			if (logger.isDebugEnabled()) {
+				logger.debug("Closing " + this + " due to remote " + status);
+			}
+
+			closeInternal(status);
+		}
+
+
+		public List<WebSocketMessage> getSentMessages() {
+			return new ArrayList<>(this.sentMessages);
+		}
+
+		@Override
+		public Mono<Void> send(Publisher<WebSocketMessage> messages) {
+			return Flux.from(messages)
+					.doOnNext(this::saveMessage)
+					.doOnNext(message -> {
+						Sinks.EmitResult result = this.sendSink.tryEmitNext(message);
+						Assert.state(result.isSuccess(), this + " failed to send: " + message + ", with " + result);
+					})
+					.then();
+		}
+
+		private void saveMessage(WebSocketMessage message) {
+			DataBuffer payload = message.getPayload().retainedSlice(0, message.getPayload().readableByteCount());
+			this.sentMessages.add(new WebSocketMessage(message.getType(), payload));
+		}
+
+		@Override
+		public Flux<WebSocketMessage> receive() {
+			return this.receiveFlux;
+		}
+
+		@Override
+		public boolean isOpen() {
+			return !Boolean.TRUE.equals(this.closeStatusSink.scan(Scannable.Attr.TERMINATED));
+		}
+
+		@Override
+		public Mono<CloseStatus> closeStatus() {
+			return this.closeStatusSink.asMono();
+		}
+
+		public Mono<Void> close(CloseStatus status) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Closing " + this + " with " + status);
+			}
+			closeInternal(status);
+			return Mono.empty();
+		}
+
+		private void closeInternal(@Nullable CloseStatus status) {
+			if (status != null) {
+				this.closeStatusSink.tryEmitValue(status);
+			}
+			else {
+				this.closeStatusSink.tryEmitEmpty();
+			};
+		}
+
+		@Override
+		public String toString() {
+			return getId();
+		}
+
+	}
+
+}