Introduce GraphQlClientException hierarchy

Refine exception handling and ensure a hierarchy of exceptions that
allows differentiating between transport errors vs field errors while
accessing an invalid data or field.

See gh-10

samples/webflux-security/src/test/java/io/spring/sample/graphql/WebFluxSecuritySampleTests.java

@@ -127,7 +127,10 @@ class WebFluxSecuritySampleTests {
 						.build()
 						.documentName("employeesNamesAndSalaries")
 						.executeAndVerify())
-				.hasMessage("Invalid handshake response getStatus: 401 Unauthorized");
+				.hasMessage(
+						"GraphQlTransport error: Invalid handshake response getStatus: 401 Unauthorized; " +
+								"nested exception is io.netty.handler.codec.http.websocketx.WebSocketClientHandshakeException: " +
+								"Invalid handshake response getStatus: 401 Unauthorized");
 	}
 
 }

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

@@ -30,10 +30,10 @@ import graphql.GraphQLError;
 
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.ResolvableType;
+import org.springframework.graphql.GraphQlRequest;
 import org.springframework.graphql.GraphQlResponse;
 import org.springframework.graphql.support.MapGraphQlResponse;
 import org.springframework.lang.Nullable;
-import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
@@ -46,45 +46,44 @@ import org.springframework.util.StringUtils;
  */
 class DefaultClientGraphQlResponse extends MapGraphQlResponse implements ClientGraphQlResponse {
 
+	private final GraphQlRequest request;
+
 	private final DocumentContext jsonPathDoc;
 
 
-	DefaultClientGraphQlResponse(GraphQlResponse response, Configuration jsonPathConfig) {
+	DefaultClientGraphQlResponse(GraphQlRequest request, GraphQlResponse response, Configuration jsonPathConfig) {
 		super(response.toMap());
+		this.request = request;
 		this.jsonPathDoc = JsonPath.parse(response.toMap(), jsonPathConfig);
 	}
 
 
+	@Override
+	public ResponseField field(String path) {
+		path = "$.data" + (StringUtils.hasText(path) ? "." + path : "");
+		return new DefaultField(this.request, this, path, this.jsonPathDoc, getErrors());
+	}
+
 	@Override
 	public <D> D toEntity(Class<D> type) {
-		assertValidResponse();
 		return field("").toEntity(type);
 	}
 
 	@Override
 	public <D> D toEntity(ParameterizedTypeReference<D> type) {
-		assertValidResponse();
 		return field("").toEntity(type);
 	}
 
-	@Override
-	public ResponseField field(String path) {
-		path = "$.data" + (StringUtils.hasText(path) ? "." + path : "");
-		return new DefaultField(path, this.jsonPathDoc, getErrors());
-	}
-
-	private void assertValidResponse() {
-		if (!isValid()) {
-			throw new IllegalStateException("Path not present exception");
-		}
-	}
-
 
 	/**
 	 * Default implementation of {@link ResponseField}.
 	 */
 	private static class DefaultField implements ResponseField {
 
+		private final GraphQlRequest request;
+
+		private final ClientGraphQlResponse response;
+
 		private final String path;
 
 		private final DocumentContext jsonPathDoc;
@@ -93,36 +92,49 @@ class DefaultClientGraphQlResponse extends MapGraphQlResponse implements ClientG
 
 		private final List<GraphQLError> errorsBelow;
 
+		private final List<GraphQLError> errorsAtOrBelow;
+
 		private final boolean exists;
 
 		@Nullable
 		private final Object value;
 
-		public DefaultField(String path, DocumentContext jsonPathDoc, List<GraphQLError> errors) {
-			Assert.notNull(path, "'path' is required");
-			this.path = path;
+		public DefaultField(
+				GraphQlRequest request, ClientGraphQlResponse response,
+				String path, DocumentContext jsonPathDoc, List<GraphQLError> errors) {
+
+			this.request = request;
+			this.response = response;
+			this.path = path ;
 			this.jsonPathDoc = jsonPathDoc;
 
+
 			List<GraphQLError> errorsAt = null;
 			List<GraphQLError> errorsBelow = null;
+			List<GraphQLError> errorsAtOrBelow = null;
 
 			for (GraphQLError error : errors) {
 				String errorPath = toJsonPath(error);
 				if (errorPath == null) {
 					continue;
 				}
-				if (errorPath.equals(path)) {
-					errorsAt = (errorsAt != null ? errorsAt : new ArrayList<>());
-					errorsAt.add(error);
-				}
 				if (errorPath.startsWith(path)) {
-					errorsBelow = (errorsBelow != null ? errorsBelow : new ArrayList<>());
-					errorsBelow.add(error);
+					if (errorPath.length() == path.length()) {
+						errorsAt = (errorsAt != null ? errorsAt : new ArrayList<>());
+						errorsAt.add(error);
+					}
+					else {
+						errorsBelow = (errorsBelow != null ? errorsBelow : new ArrayList<>());
+						errorsBelow.add(error);
+					}
+					errorsAtOrBelow = (errorsAtOrBelow != null ? errorsAtOrBelow : new ArrayList<>());
+					errorsAtOrBelow.add(error);
 				}
 			}
 
 			this.errorsAt = (errorsAt != null ? errorsAt : Collections.emptyList());
 			this.errorsBelow = (errorsBelow != null ? errorsBelow : Collections.emptyList());
+			this.errorsAtOrBelow = (errorsAtOrBelow != null ? errorsAtOrBelow : Collections.emptyList());
 
 
 			boolean exists = true;
@@ -183,35 +195,38 @@ class DefaultClientGraphQlResponse extends MapGraphQlResponse implements ClientG
 			return this.errorsBelow;
 		}
 
+		@Override
+		public List<GraphQLError> getErrorsAtOrBelow() {
+			return this.errorsAtOrBelow;
+		}
+
 		@Override
 		public <D> D toEntity(Class<D> entityType) {
-			assertValidField();
+			assertIsValid();
 			return this.jsonPathDoc.read(this.path, new TypeRefAdapter<>(entityType));
 		}
 
 		@Override
 		public <D> D toEntity(ParameterizedTypeReference<D> entityType) {
-			assertValidField();
+			assertIsValid();
 			return this.jsonPathDoc.read(this.path, new TypeRefAdapter<>(entityType));
 		}
 
 		@Override
 		public <D> List<D> toEntityList(Class<D> elementType) {
-			assertValidField();
+			assertIsValid();
 			return this.jsonPathDoc.read(this.path, new TypeRefAdapter<>(List.class, elementType));
 		}
 
 		@Override
 		public <D> List<D> toEntityList(ParameterizedTypeReference<D> elementType) {
-			assertValidField();
+			assertIsValid();
 			return this.jsonPathDoc.read(this.path, new TypeRefAdapter<>(List.class, elementType));
 		}
 
-		private void assertValidField() {
+		private void assertIsValid() {
 			if (!isValid()) {
-				throw (CollectionUtils.isEmpty(this.errorsAt) ?
-						new IllegalStateException("Path not present exception") :
-						new IllegalStateException("Field error exception"));
+				throw new FieldAccessException(this.request, this.response, this);
 			}
 		}
 

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

@@ -145,14 +145,22 @@ final class DefaultGraphQlClient implements GraphQlClient {
 		public Mono<ClientGraphQlResponse> execute() {
 			return initRequest().flatMap(request ->
 					this.transport.execute(request)
-							.map(response -> new DefaultClientGraphQlResponse(response, this.jsonPathConfig)));
+							.map(result ->
+									new DefaultClientGraphQlResponse(request, result, this.jsonPathConfig))
+							.onErrorResume(
+									ex -> !(ex instanceof GraphQlClientException),
+									ex -> toGraphQlTransportException(ex, request)));
 		}
 
 		@Override
 		public Flux<ClientGraphQlResponse> executeSubscription() {
 			return initRequest().flatMapMany(request ->
 					this.transport.executeSubscription(request)
-							.map(response -> new DefaultClientGraphQlResponse(response, this.jsonPathConfig)));
+							.map(result ->
+									new DefaultClientGraphQlResponse(request, result, this.jsonPathConfig))
+							.onErrorResume(
+									ex -> !(ex instanceof GraphQlClientException),
+									ex -> toGraphQlTransportException(ex, request)));
 		}
 
 		private Mono<GraphQlRequest> initRequest() {
@@ -160,6 +168,10 @@ final class DefaultGraphQlClient implements GraphQlClient {
 					new GraphQlRequest(document, this.operationName, this.variables));
 		}
 
+		private <T> Mono<T> toGraphQlTransportException(Throwable ex, GraphQlRequest request) {
+			return Mono.error(new GraphQlTransportException(ex, request));
+		}
+
 	}
 
 

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

@@ -0,0 +1,67 @@
+/*
+ * 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 org.springframework.graphql.GraphQlRequest;
+import org.springframework.graphql.GraphQlResponse;
+
+/**
+ * An exception raised on an attempt to decode data from an
+ * {@link GraphQlResponse#isValid() invalid response} or an
+ * {@link ResponseField#isValid() invalid field}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+@SuppressWarnings("serial")
+public class FieldAccessException extends GraphQlClientException {
+
+	private final ClientGraphQlResponse response;
+
+	private final ResponseField field;
+
+
+	/**
+	 * Constructor with the request and response, and the accessed field.
+	 */
+	public FieldAccessException(GraphQlRequest request, ClientGraphQlResponse response, ResponseField field) {
+		super(initDefaultMessage(field), null, request);
+		this.response = response;
+		this.field = field;
+	}
+
+	private static String initDefaultMessage(ResponseField field) {
+		return "Invalid field '" + field.getPath() + "', errors: " + field.getErrorsAtOrBelow();
+	}
+
+
+	/**
+	 * Return the [@code GraphQlResponse} for which the error ouccrred.
+	 */
+	public ClientGraphQlResponse getResponse() {
+		return this.response;
+	}
+
+	/**
+	 * Return the field that needed to be accessed.
+	 */
+	public ResponseField getField() {
+		return this.field;
+	}
+
+}

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

@@ -0,0 +1,53 @@
+/*
+ * 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 org.springframework.core.NestedRuntimeException;
+import org.springframework.graphql.GraphQlRequest;
+import org.springframework.lang.Nullable;
+
+
+/**
+ * Base class for exceptions from {@code GraphQlClient}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+@SuppressWarnings("serial")
+public class GraphQlClientException extends NestedRuntimeException {
+
+	private final GraphQlRequest request;
+
+
+	/**
+	 * Constructor with a message, optional cause, and the request details.
+	 */
+	public GraphQlClientException(String message, @Nullable Throwable cause, GraphQlRequest request) {
+		super(message, cause);
+		this.request = request;
+	}
+
+
+	/**
+	 * Return the request for which the error occurred.
+	 */
+	public GraphQlRequest getRequest() {
+		return this.request;
+	}
+
+}

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

@@ -0,0 +1,47 @@
+/*
+ * 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 org.springframework.graphql.GraphQlRequest;
+import org.springframework.lang.Nullable;
+
+/**
+ * Exception raised by a {@link GraphQlTransport} or used to wrap an exception
+ * from a {@code GraphQlTransport} implementation.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+@SuppressWarnings("serial")
+public class GraphQlTransportException extends GraphQlClientException {
+
+	/**
+	 * Constructor with a default message.
+	 */
+	public GraphQlTransportException(@Nullable Throwable cause, GraphQlRequest request) {
+		super("GraphQlTransport error: " + cause.getMessage(), cause, request);
+	}
+
+	/**
+	 * Constructor with a given message.
+	 */
+	public GraphQlTransportException(String message, @Nullable Throwable cause, GraphQlRequest request) {
+		super(message, cause, request);
+	}
+
+}

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

@@ -69,6 +69,11 @@ public interface ResponseField {
 	 */
 	List<GraphQLError> getErrorsBelow();
 
+	/**
+	 * Return errors with paths at or below that of the field.
+	 */
+	List<GraphQLError> getErrorsAtOrBelow();
+
 	/**
 	 * Decode the field to an entity of the given type.
 	 * @param entityType the type to convert to

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

@@ -20,22 +20,29 @@ import java.util.List;
 
 import graphql.GraphQLError;
 
+import org.springframework.graphql.GraphQlRequest;
+
 /**
- * Exception that is sent as an error signal to a {@code Flux} returned from
- * {@link GraphQlClient} or from its underlying {@link GraphQlTransport} for a
- * GraphQL over WebSocket subscription that ends with an "error" message.
+ * WebSocket {@link GraphQlTransportException} raised when a subscription
+ * ends with an {@code "error"} message. The {@link #getErrors()} method provides
+ * access to the GraphQL errors from the message payload.
  *
  * @author Rossen Stoyanchev
  * @since 1.0.0
  */
 @SuppressWarnings("serial")
-public class SubscriptionErrorException extends RuntimeException {
+public class SubscriptionErrorException extends GraphQlTransportException {
 
 	private final List<GraphQLError> errors;
 
 
-	public SubscriptionErrorException(List<GraphQLError> errors) {
-		super("GraphQL subscription error: " + errors);
+	/**
+	 * Constructor with the request details and the errors listed in the payload
+	 * of the {@code "errors"} message.
+	 */
+	public SubscriptionErrorException(GraphQlRequest request, List<GraphQLError> errors) {
+		super("GraphQL subscription completed with an \"error\" message, " +
+				"with the following errors: " + errors, null, request);
 		this.errors = errors;
 	}
 

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

@@ -0,0 +1,52 @@
+/*
+ * 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 org.springframework.graphql.GraphQlRequest;
+import org.springframework.web.reactive.socket.CloseStatus;
+
+/**
+ * WebSocket related {@link GraphQlTransportException} raised when the connection
+ * is closed while a request or subscription is in progress.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+@SuppressWarnings("serial")
+public class WebSocketDisconnectedException extends GraphQlTransportException {
+
+	private final CloseStatus closeStatus;
+
+
+	/**
+	 * Constructor with an explanation about the closure, along with the request
+	 * details and the status used to close the WebSocket session.
+	 */
+	public WebSocketDisconnectedException(String closeStatusMessage, GraphQlRequest request, CloseStatus status) {
+		super(closeStatusMessage, null, request);
+		this.closeStatus = status;
+	}
+
+
+	/**
+	 * Return the {@link CloseStatus} used to close the WebSocket session.
+	 */
+	public CloseStatus getCloseStatus() {
+		return this.closeStatus;
+	}
+
+}

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

@@ -260,22 +260,31 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 							}
 						}
 						else {
-							GraphQlMessage message = this.codecDelegate.decode(webSocketMessage);
-							switch (message.resolvedType()) {
-								case NEXT:
-									graphQlSession.handleNext(message);
-									break;
-								case PING:
-									graphQlSession.sendPong(null);
-									break;
-								case ERROR:
-									graphQlSession.handleError(message);
-									break;
-								case COMPLETE:
-									graphQlSession.handleComplete(message);
-									break;
-								default:
-									return session.close(new CloseStatus(4400, "Invalid message"));
+							try {
+								GraphQlMessage message = this.codecDelegate.decode(webSocketMessage);
+								switch (message.resolvedType()) {
+									case NEXT:
+										graphQlSession.handleNext(message);
+										break;
+									case PING:
+										graphQlSession.sendPong(null);
+										break;
+									case ERROR:
+										graphQlSession.handleError(message);
+										break;
+									case COMPLETE:
+										graphQlSession.handleComplete(message);
+										break;
+									default:
+										throw new IllegalStateException(
+												"Unexpected message type: '" + message.getType() + "'");
+								}
+							}
+							catch (Exception ex) {
+								if (logger.isErrorEnabled()) {
+									logger.error("Closing " + session + ": " + ex);
+								}
+								return session.close(new CloseStatus(4400, "Invalid message"));
 							}
 						}
 						return Mono.empty();
@@ -293,18 +302,19 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 			session.closeStatus()
 					.defaultIfEmpty(CloseStatus.NO_STATUS_CODE)
 					.doOnNext(closeStatus -> {
-						Exception ex = initDisconnectError(closeStatus, null, graphQlSession);
+						String closeStatusMessage = initCloseStatusMessage(closeStatus, null, graphQlSession);
 						if (logger.isDebugEnabled()) {
-							logger.debug(ex.getMessage());
+							logger.debug(closeStatusMessage);
 						}
-						graphQlSession.terminateRequests(ex);
+						graphQlSession.terminateRequests(closeStatusMessage, closeStatus);
 					})
 					.doOnError(cause -> {
-						Exception ex = initDisconnectError(null, cause, graphQlSession);
+						CloseStatus closeStatus = CloseStatus.NO_STATUS_CODE;
+						String closeStatusMessage = initCloseStatusMessage(closeStatus, cause, graphQlSession);
 						if (logger.isErrorEnabled()) {
-							logger.error(ex.getMessage());
+							logger.error(closeStatusMessage);
 						}
-						graphQlSession.terminateRequests(ex);
+						graphQlSession.terminateRequests(closeStatusMessage, closeStatus);
 					})
 					.doOnTerminate(() -> {
 						// Reset GraphQlSession sink to be ready to connect again
@@ -313,23 +323,21 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 					.subscribe();
 		}
 
-		private Exception initDisconnectError(
-				@Nullable CloseStatus status, @Nullable Throwable ex, GraphQlSession graphQlSession) {
-
-			String reason = graphQlSession + " disconnected";
+		private String initCloseStatusMessage(CloseStatus status, @Nullable Throwable ex, GraphQlSession session) {
+			String reason = session + " disconnected";
 			if (isStopped()) {
-				reason = graphQlSession + " was stopped";
+				reason = session + " was stopped";
 			}
 			else if (ex != null) {
 				reason += ", closeStatus() completed with error " + ex;
 			}
-			else if (status != null && !status.equals(CloseStatus.NO_STATUS_CODE)) {
+			else if (!status.equals(CloseStatus.NO_STATUS_CODE)) {
 				reason += " with " + status;
 			}
 			else {
 				reason += " without a status";
 			}
-			return new IllegalStateException(reason);
+			return reason;
 		}
 
 		/**
@@ -372,9 +380,9 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 
 		private final Sinks.Many<GraphQlMessage> requestSink = Sinks.many().unicast().onBackpressureBuffer();
 
-		private final Map<String, Sinks.One<GraphQlResponse>> responseSinks = new ConcurrentHashMap<>();
+		private final Map<String, ResponseState> responseMap = new ConcurrentHashMap<>();
 
-		private final Map<String, Sinks.Many<GraphQlResponse>> streamSinks = new ConcurrentHashMap<>();
+		private final Map<String, SubscriptionState> subscriptionMap = new ConcurrentHashMap<>();
 
 
 		GraphQlSession(WebSocketSession webSocketSession) {
@@ -393,13 +401,13 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 			String id = String.valueOf(this.requestIndex.incrementAndGet());
 			try {
 				GraphQlMessage message = GraphQlMessage.subscribe(id, request);
-				Sinks.One<GraphQlResponse> sink = Sinks.one();
-				this.responseSinks.put(id, sink);
+				ResponseState state = new ResponseState(request);
+				this.responseMap.put(id, state);
 				trySend(message);
-				return sink.asMono().doOnCancel(() -> this.responseSinks.remove(id));
+				return state.sink().asMono().doOnCancel(() -> this.responseMap.remove(id));
 			}
 			catch (Exception ex) {
-				this.responseSinks.remove(id);
+				this.responseMap.remove(id);
 				return Mono.error(ex);
 			}
 		}
@@ -408,13 +416,13 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 			String id = String.valueOf(this.requestIndex.incrementAndGet());
 			try {
 				GraphQlMessage message = GraphQlMessage.subscribe(id, request);
-				Sinks.Many<GraphQlResponse> sink = Sinks.many().unicast().onBackpressureBuffer();
-				this.streamSinks.put(id, sink);
+				SubscriptionState state = new SubscriptionState(request);
+				this.subscriptionMap.put(id, state);
 				trySend(message);
-				return sink.asFlux().doOnCancel(() -> cancelStream(id));
+				return state.sink().asFlux().doOnCancel(() -> stopSubscription(id));
 			}
 			catch (Exception ex) {
-				this.streamSinks.remove(id);
+				this.subscriptionMap.remove(id);
 				return Flux.error(ex);
 			}
 		}
@@ -437,9 +445,9 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 			Assert.state(emitResult.isSuccess(), "Failed to send request: " + emitResult);
 		}
 
-		private void cancelStream(String id) {
-			Sinks.Many<GraphQlResponse> streamSink = this.streamSinks.remove(id);
-			if (streamSink != null) {
+		private void stopSubscription(String id) {
+			SubscriptionState state = this.subscriptionMap.remove(id);
+			if (state != null) {
 				try {
 					trySend(GraphQlMessage.complete(id));
 				}
@@ -459,10 +467,10 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 		 */
 		public void handleNext(GraphQlMessage message) {
 			String id = message.getId();
-			Sinks.One<GraphQlResponse> sink = this.responseSinks.remove(id);
-			Sinks.Many<GraphQlResponse> streamingSink = this.streamSinks.get(id);
+			ResponseState responseState = this.responseMap.remove(id);
+			SubscriptionState subscriptionState = this.subscriptionMap.get(id);
 
-			if (sink == null && streamingSink == null) {
+			if (responseState == null && subscriptionState == null) {
 				if (logger.isDebugEnabled()) {
 					logger.debug("No receiver for message: " + message);
 				}
@@ -470,9 +478,12 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 			}
 
 			Map<String, Object> responseMap = message.getPayload();
-			GraphQlResponse response = MapGraphQlResponse.forResponse(responseMap);
+			GraphQlResponse graphQlResponse = MapGraphQlResponse.forResponse(responseMap);
+
+			Sinks.EmitResult emitResult = (responseState != null ?
+					responseState.sink().tryEmitValue(graphQlResponse) :
+					subscriptionState.sink().tryEmitNext(graphQlResponse));
 
-			Sinks.EmitResult emitResult = (sink != null ? sink.tryEmitValue(response) : streamingSink.tryEmitNext(response));
 			if (emitResult.isFailure()) {
 				// Just log: cannot overflow, is serialized, and cancel is handled in doOnCancel
 				if (logger.isDebugEnabled()) {
@@ -487,10 +498,10 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 		 */
 		public void handleError(GraphQlMessage message) {
 			String id = message.getId();
-			Sinks.One<GraphQlResponse> sink = this.responseSinks.remove(id);
-			Sinks.Many<GraphQlResponse> streamingSink = this.streamSinks.remove(id);
+			ResponseState responseState = this.responseMap.remove(id);
+			SubscriptionState subscriptionState = this.subscriptionMap.remove(id);
 
-			if (sink == null && streamingSink == null ) {
+			if (responseState == null && subscriptionState == null) {
 				if (logger.isDebugEnabled()) {
 					logger.debug("No receiver for message: " + message);
 				}
@@ -500,14 +511,14 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 			List<Map<String, Object>> errorList = message.getPayload();
 
 			Sinks.EmitResult emitResult;
-			if (sink != null) {
+			if (responseState != null) {
 				GraphQlResponse response = MapGraphQlResponse.forErrorsOnly(errorList);
-				emitResult = sink.tryEmitValue(response);
+				emitResult = responseState.sink().tryEmitValue(response);
 			}
 			else {
 				List<GraphQLError> graphQLErrors = MapGraphQlError.from(errorList);
-				Exception ex = new SubscriptionErrorException(graphQLErrors);
-				emitResult = streamingSink.tryEmitError(ex);
+				Exception ex = new SubscriptionErrorException(subscriptionState.request(), graphQLErrors);
+				emitResult = subscriptionState.sink().tryEmitError(ex);
 			}
 
 			if (emitResult.isFailure() && logger.isDebugEnabled()) {
@@ -519,14 +530,14 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 		 * Handle a "complete" message.
 		 */
 		public void handleComplete(GraphQlMessage message) {
-			Sinks.One<GraphQlResponse> sink = this.responseSinks.remove(message.getId());
-			Sinks.Many<GraphQlResponse> streamSink = this.streamSinks.remove(message.getId());
+			ResponseState responseState = this.responseMap.remove(message.getId());
+			SubscriptionState subscriptionState = this.subscriptionMap.remove(message.getId());
 
-			if (sink != null) {
-				sink.tryEmitEmpty();
+			if (responseState != null) {
+				responseState.sink().tryEmitEmpty();
 			}
-			else if (streamSink != null) {
-				streamSink.tryEmitComplete();
+			else if (subscriptionState != null) {
+				subscriptionState.sink().tryEmitComplete();
 			}
 		}
 
@@ -548,11 +559,11 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 		/**
 		 * Terminate and clean all in-progress requests with the given error.
 		 */
-		public void terminateRequests(Exception ex) {
-			this.responseSinks.values().forEach(sink -> sink.tryEmitError(ex));
-			this.streamSinks.values().forEach(sink -> sink.tryEmitError(ex));
-			this.responseSinks.clear();
-			this.streamSinks.clear();
+		public void terminateRequests(String message, CloseStatus status) {
+			this.responseMap.values().forEach(info -> info.emitDisconnectError(message, status));
+			this.subscriptionMap.values().forEach(info -> info.emitDisconnectError(message, status)			);
+			this.responseMap.clear();
+			this.subscriptionMap.clear();
 		}
 
 		@Override
@@ -600,4 +611,73 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
 	}
 
 
+	/**
+	 * Base class, state container for any request type.
+	 */
+	private abstract static class AbstractRequestState {
+
+		private final GraphQlRequest request;
+
+		public AbstractRequestState(GraphQlRequest request) {
+			this.request = request;
+		}
+
+		public GraphQlRequest request() {
+			return this.request;
+		}
+
+		public void emitDisconnectError(String message, CloseStatus closeStatus) {
+			emitDisconnectError(new WebSocketDisconnectedException(message, this.request, closeStatus));
+		}
+
+		protected abstract void emitDisconnectError(WebSocketDisconnectedException ex);
+
+	}
+
+
+	/**
+	 * State container for a request that emits a single response.
+	 */
+	private static class ResponseState extends AbstractRequestState {
+
+		private final Sinks.One<GraphQlResponse> sink = Sinks.one();
+
+		ResponseState(GraphQlRequest request) {
+			super(request);
+		}
+
+		public Sinks.One<GraphQlResponse> sink() {
+			return this.sink;
+		}
+
+		@Override
+		protected void emitDisconnectError(WebSocketDisconnectedException ex) {
+			this.sink.tryEmitError(ex);
+		}
+
+	}
+
+
+	/**
+	 * State container for a subscription request that emits a stream of responses.
+	 */
+	private static class SubscriptionState extends AbstractRequestState {
+
+		private final Sinks.Many<GraphQlResponse> sink = Sinks.many().unicast().onBackpressureBuffer();
+
+		SubscriptionState(GraphQlRequest request) {
+			super(request);
+		}
+
+		public Sinks.Many<GraphQlResponse> sink() {
+			return this.sink;
+		}
+
+		@Override
+		protected void emitDisconnectError(WebSocketDisconnectedException ex) {
+			this.sink.tryEmitError(ex);
+		}
+
+	}
+
 }