diff --git a/samples/webflux-security/src/test/java/io/spring/sample/graphql/WebFluxSecuritySampleTests.java b/samples/webflux-security/src/test/java/io/spring/sample/graphql/WebFluxSecuritySampleTests.java index 6336f258..cffee115 100644 --- a/samples/webflux-security/src/test/java/io/spring/sample/graphql/WebFluxSecuritySampleTests.java +++ b/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"); } } \ No newline at end of file diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultClientGraphQlResponse.java b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultClientGraphQlResponse.java index 1d9cf80e..bb080a66 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultClientGraphQlResponse.java +++ b/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 toEntity(Class type) { - assertValidResponse(); return field("").toEntity(type); } @Override public D toEntity(ParameterizedTypeReference 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 errorsBelow; + private final List errorsAtOrBelow; + private final boolean exists; @Nullable private final Object value; - public DefaultField(String path, DocumentContext jsonPathDoc, List errors) { - Assert.notNull(path, "'path' is required"); - this.path = path; + public DefaultField( + GraphQlRequest request, ClientGraphQlResponse response, + String path, DocumentContext jsonPathDoc, List errors) { + + this.request = request; + this.response = response; + this.path = path ; this.jsonPathDoc = jsonPathDoc; + List errorsAt = null; List errorsBelow = null; + List 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 getErrorsAtOrBelow() { + return this.errorsAtOrBelow; + } + @Override public D toEntity(Class entityType) { - assertValidField(); + assertIsValid(); return this.jsonPathDoc.read(this.path, new TypeRefAdapter<>(entityType)); } @Override public D toEntity(ParameterizedTypeReference entityType) { - assertValidField(); + assertIsValid(); return this.jsonPathDoc.read(this.path, new TypeRefAdapter<>(entityType)); } @Override public List toEntityList(Class elementType) { - assertValidField(); + assertIsValid(); return this.jsonPathDoc.read(this.path, new TypeRefAdapter<>(List.class, elementType)); } @Override public List toEntityList(ParameterizedTypeReference 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); } } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultGraphQlClient.java b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultGraphQlClient.java index 8839ae7b..1d0414f4 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultGraphQlClient.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultGraphQlClient.java @@ -145,14 +145,22 @@ final class DefaultGraphQlClient implements GraphQlClient { public Mono 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 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 initRequest() { @@ -160,6 +168,10 @@ final class DefaultGraphQlClient implements GraphQlClient { new GraphQlRequest(document, this.operationName, this.variables)); } + private Mono toGraphQlTransportException(Throwable ex, GraphQlRequest request) { + return Mono.error(new GraphQlTransportException(ex, request)); + } + } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/FieldAccessException.java b/spring-graphql/src/main/java/org/springframework/graphql/client/FieldAccessException.java new file mode 100644 index 00000000..284cae95 --- /dev/null +++ b/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; + } + +} diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/GraphQlClientException.java b/spring-graphql/src/main/java/org/springframework/graphql/client/GraphQlClientException.java new file mode 100644 index 00000000..543ad296 --- /dev/null +++ b/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; + } + +} diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/GraphQlTransportException.java b/spring-graphql/src/main/java/org/springframework/graphql/client/GraphQlTransportException.java new file mode 100644 index 00000000..8dca147e --- /dev/null +++ b/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); + } + +} diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseField.java b/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseField.java index c941435c..9a4e4751 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseField.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseField.java @@ -69,6 +69,11 @@ public interface ResponseField { */ List getErrorsBelow(); + /** + * Return errors with paths at or below that of the field. + */ + List getErrorsAtOrBelow(); + /** * Decode the field to an entity of the given type. * @param entityType the type to convert to diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/SubscriptionErrorException.java b/spring-graphql/src/main/java/org/springframework/graphql/client/SubscriptionErrorException.java index 69b52be6..b83654d2 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/SubscriptionErrorException.java +++ b/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 errors; - public SubscriptionErrorException(List 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 errors) { + super("GraphQL subscription completed with an \"error\" message, " + + "with the following errors: " + errors, null, request); this.errors = errors; } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketDisconnectedException.java b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketDisconnectedException.java new file mode 100644 index 00000000..eb64ef8b --- /dev/null +++ b/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; + } + +} diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlTransport.java b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlTransport.java index d3e9bdf9..5e7405b7 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlTransport.java +++ b/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 requestSink = Sinks.many().unicast().onBackpressureBuffer(); - private final Map> responseSinks = new ConcurrentHashMap<>(); + private final Map responseMap = new ConcurrentHashMap<>(); - private final Map> streamSinks = new ConcurrentHashMap<>(); + private final Map 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 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 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 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 sink = this.responseSinks.remove(id); - Sinks.Many 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 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 sink = this.responseSinks.remove(id); - Sinks.Many 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> 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 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 sink = this.responseSinks.remove(message.getId()); - Sinks.Many 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 sink = Sinks.one(); + + ResponseState(GraphQlRequest request) { + super(request); + } + + public Sinks.One 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 sink = Sinks.many().unicast().onBackpressureBuffer(); + + SubscriptionState(GraphQlRequest request) { + super(request); + } + + public Sinks.Many sink() { + return this.sink; + } + + @Override + protected void emitDisconnectError(WebSocketDisconnectedException ex) { + this.sink.tryEmitError(ex); + } + + } + }