Spring/spring-graphql
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing in WebSocket support

Browse Source
This commit is contained in:
rstoyanchev
2023-04-18 09:30:34 +01:00
parent 84f8cfe446
commit 3b1d1abfb9
8 changed files with 68 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -132,7 +132,7 @@ final class DefaultWebSocketGraphQlClientBuilder
List<WebSocketGraphQlClientInterceptor> interceptors = getInterceptors().stream()
.filter(interceptor -> interceptor instanceof WebSocketGraphQlClientInterceptor)
.map(interceptor -> (WebSocketGraphQlClientInterceptor) interceptor)
.collect(Collectors.toList());
.toList();
Assert.state(interceptors.size() <= 1,
"Only a single interceptor of type WebSocketGraphQlClientInterceptor may be configured");

6
spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClientInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2023 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.
@@ -27,6 +27,10 @@ import reactor.core.publisher.Mono;
* for WebSocket interception points. Only a single interceptor of type
* {@link WebSocketGraphQlClientInterceptor} may be configured.
*
* <p>Use {@link GraphQlClient.Builder#interceptor(GraphQlClientInterceptor...)}
* to configure the interceptor chain. Only one interceptor in the chain may be
* of type {@code WebSocketGraphQlClientInterceptor}.
*
* @author Rossen Stoyanchev
* @since 1.0.0
*/

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -87,6 +87,7 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
.cacheInvalidateWhen(GraphQlSession::notifyWhenClosed);
}
@SuppressWarnings({"CallingSubscribeInNonBlockingScope", "ReactorTransformationOnMonoVoid"})
private static Mono<GraphQlSession> initGraphQlSession(
URI uri, HttpHeaders headers, WebSocketClient client, GraphQlSessionHandler handler) {
@@ -219,6 +220,7 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
}
@SuppressWarnings({"ReactorZipWithMonoVoid", "ReactiveStreamsThrowInOperator"})
@Override
public Mono<Void> handle(WebSocketSession session) {
@@ -265,21 +267,12 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
try {
GraphQlWebSocketMessage 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() + "'");
case NEXT -> graphQlSession.handleNext(message);
case PING -> graphQlSession.sendPong(null);
case ERROR -> graphQlSession.handleError(message);
case COMPLETE -> graphQlSession.handleComplete(message);
default -> throw new IllegalStateException(
"Unexpected message type: '" + message.getType() + "'");
}
}
catch (Exception ex) {
@@ -501,7 +494,7 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
}
else {
List<ResponseError> errors = response.getErrors();
Exception ex = new SubscriptionErrorException(requestState.getRequest(), errors);
Exception ex = new SubscriptionErrorException(requestState.request(), errors);
requestState.handlerError(ex);
}
}
@@ -619,7 +612,7 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
*/
private interface RequestState {
GraphQlRequest getRequest();
GraphQlRequest request();
void handleResponse(GraphQlResponse response);
@@ -628,7 +621,7 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
void handleCompletion();
default void emitDisconnectError(String message, CloseStatus closeStatus) {
emitDisconnectError(new WebSocketDisconnectedException(message, getRequest(), closeStatus));
emitDisconnectError(new WebSocketDisconnectedException(message, request(), closeStatus));
}
void emitDisconnectError(WebSocketDisconnectedException ex);
@@ -639,21 +632,8 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
/**
* State container for a request that emits a single response.
*/
private static class SingleResponseRequestState implements RequestState {
private final GraphQlRequest request;
private final MonoSink<GraphQlResponse> responseSink;
SingleResponseRequestState(GraphQlRequest request, MonoSink<GraphQlResponse> responseSink) {
this.request = request;
this.responseSink = responseSink;
}
@Override
public GraphQlRequest getRequest() {
return this.request;
}
private record SingleResponseRequestState(
GraphQlRequest request, MonoSink<GraphQlResponse> responseSink) implements RequestState {
@Override
public void handleResponse(GraphQlResponse response) {
@@ -681,21 +661,8 @@ final class WebSocketGraphQlTransport implements GraphQlTransport {
/**
* State container for a subscription request that emits a stream of responses.
*/
private static class SubscriptionRequestState implements RequestState {
private final GraphQlRequest request;
private final FluxSink<GraphQlResponse> responseSink;
SubscriptionRequestState(GraphQlRequest request, FluxSink<GraphQlResponse> responseSink) {
this.request = request;
this.responseSink = responseSink;
}
@Override
public GraphQlRequest getRequest() {
return request;
}
private record SubscriptionRequestState(
GraphQlRequest request, FluxSink<GraphQlResponse> responseSink) implements RequestState {
@Override
public void handleResponse(GraphQlResponse response) {

14
spring-graphql/src/main/java/org/springframework/graphql/server/WebSocketGraphQlInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -22,9 +22,11 @@ import reactor.core.publisher.Mono;
/**
* An extension of {@link WebGraphQlInterceptor} with additional methods
* to handle the start and end of a WebSocket connection. Only a single
* interceptor of type {@link WebSocketGraphQlInterceptor} may be
* declared.
* to handle the start and end of a WebSocket connection.
*
* <p>Use {@link WebGraphQlHandler.Builder#interceptor(WebGraphQlInterceptor...)}
* to configure the interceptor chain. Only one interceptor in the chain may be
* of type {@code WebSocketGraphQlInterceptor}.
*
* @author Rossen Stoyanchev
* @since 1.0.0
@@ -57,8 +59,8 @@ public interface WebSocketGraphQlInterceptor extends WebGraphQlInterceptor {
* additional, or more centralized handling across subscriptions.
* @param sessionInfo information about the underlying WebSocket session
* @param subscriptionId the unique id for the subscription; correlates to the
* {@link WebGraphQlRequest#getId() requestId} from the original {@code "subscribe"}
* message that started the subscription
* {@link WebGraphQlRequest#getId() requestId} from the {@code "subscribe"}
* message that started the subscription stream
* @return {@code Mono} for the completion of handling
*/
default Mono<Void> handleCancelledSubscription(WebSocketSessionInfo sessionInfo, String subscriptionId) {

16
spring-graphql/src/main/java/org/springframework/graphql/server/webflux/GraphQlWebSocketHandler.java
View File

@@ -101,6 +101,7 @@ public class GraphQlWebSocketHandler implements WebSocketHandler {
}
@SuppressWarnings("CallingSubscribeInNonBlockingScope")
@Override
public Mono<Void> handle(WebSocketSession session) {
HandshakeInfo handshakeInfo = session.getHandshakeInfo();
@@ -140,7 +141,7 @@ public class GraphQlWebSocketHandler implements WebSocketHandler {
String id = message.getId();
Map<String, Object> payload = message.getPayload();
switch (message.resolvedType()) {
case SUBSCRIBE:
case SUBSCRIBE -> {
if (connectionInitPayloadRef.get() == null) {
return GraphQlStatus.close(session, GraphQlStatus.UNAUTHORIZED_STATUS);
}
@@ -156,9 +157,11 @@ public class GraphQlWebSocketHandler implements WebSocketHandler {
return this.graphQlHandler.handleRequest(request)
.flatMapMany(response -> handleResponse(session, id, subscriptions, response))
.doOnTerminate(() -> subscriptions.remove(id));
case PING:
}
case PING -> {
return Flux.just(this.codecDelegate.encode(session, GraphQlWebSocketMessage.pong(null)));
case COMPLETE:
}
case COMPLETE -> {
if (id != null) {
Subscription subscription = subscriptions.remove(id);
if (subscription != null) {
@@ -168,7 +171,8 @@ public class GraphQlWebSocketHandler implements WebSocketHandler {
.thenMany(Flux.empty());
}
return Flux.empty();
case CONNECTION_INIT:
}
case CONNECTION_INIT -> {
if (!connectionInitPayloadRef.compareAndSet(null, payload)) {
return GraphQlStatus.close(session, GraphQlStatus.TOO_MANY_INIT_REQUESTS_STATUS);
}
@@ -177,8 +181,10 @@ public class GraphQlWebSocketHandler implements WebSocketHandler {
.map(ackPayload -> this.codecDelegate.encodeConnectionAck(session, ackPayload))
.flux()
.onErrorResume(ex -> GraphQlStatus.close(session, GraphQlStatus.UNAUTHORIZED_STATUS));
default:
}
default -> {
return GraphQlStatus.close(session, GraphQlStatus.INVALID_MESSAGE_STATUS);
}
}
}));
}

24
spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlWebSocketHandler.java
View File

@@ -191,7 +191,7 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub
Map<String, Object> payload = message.getPayload();
SessionState state = getSessionInfo(session);
switch (message.resolvedType()) {
case SUBSCRIBE:
case SUBSCRIBE -> {
if (state.getConnectionInitPayload() == null) {
GraphQlStatus.closeSession(session, GraphQlStatus.UNAUTHORIZED_STATUS);
return;
@@ -212,11 +212,9 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub
.flatMapMany((response) -> handleResponse(session, request.getId(), response))
.publishOn(state.getScheduler()) // Serial blocking send via single thread
.subscribe(new SendMessageSubscriber(id, session, state));
return;
case PING:
session.sendMessage(encode(GraphQlWebSocketMessage.pong(null)));
return;
case COMPLETE:
}
case PING -> session.sendMessage(encode(GraphQlWebSocketMessage.pong(null)));
case COMPLETE -> {
if (id != null) {
Subscription subscription = state.getSubscriptions().remove(id);
if (subscription != null) {
@@ -225,8 +223,8 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub
this.webSocketGraphQlInterceptor.handleCancelledSubscription(state.getSessionInfo(), id)
.block(Duration.ofSeconds(10));
}
return;
case CONNECTION_INIT:
}
case CONNECTION_INIT -> {
if (!state.setConnectionInitPayload(payload)) {
GraphQlStatus.closeSession(session, GraphQlStatus.TOO_MANY_INIT_REQUESTS_STATUS);
return;
@@ -248,9 +246,8 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub
return Mono.empty();
})
.block(Duration.ofSeconds(10));
return;
default:
GraphQlStatus.closeSession(session, GraphQlStatus.INVALID_MESSAGE_STATUS);
}
default -> GraphQlStatus.closeSession(session, GraphQlStatus.INVALID_MESSAGE_STATUS);
}
}
@@ -343,11 +340,6 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub
}
}
@Override
public boolean supportsPartialMessages() {
return false;
}
/**
* {@code HandshakeInterceptor} that propagates ThreadLocal context through

14
spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -88,7 +88,7 @@ public final class MockGraphQlWebSocketServer implements WebSocketHandler {
@SuppressWarnings("SuspiciousMethodCalls")
private Publisher<GraphQlWebSocketMessage> handleMessage(GraphQlWebSocketMessage message) {
switch (message.resolvedType()) {
case CONNECTION_INIT:
case CONNECTION_INIT -> {
if (this.connectionInitHandler == null) {
return Flux.just(GraphQlWebSocketMessage.connectionAck(null));
}
@@ -96,7 +96,8 @@ public final class MockGraphQlWebSocketServer implements WebSocketHandler {
Map<String, Object> payload = message.getPayload();
return this.connectionInitHandler.apply(payload).map(GraphQlWebSocketMessage::connectionAck);
}
case SUBSCRIBE:
}
case SUBSCRIBE -> {
String id = message.getId();
Exchange request = expectedExchanges.get(message.getPayload());
if (id == null || request == null) {
@@ -108,10 +109,13 @@ public final class MockGraphQlWebSocketServer implements WebSocketHandler {
request.getError() != null ?
GraphQlWebSocketMessage.error(id, Collections.singletonList(request.getError())) :
GraphQlWebSocketMessage.complete(id));
case COMPLETE:
}
case COMPLETE -> {
return Flux.empty();
default:
}
default -> {
return Flux.error(new IllegalStateException("Unexpected message: " + message));
}
}
}

22
spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -360,18 +360,14 @@ public class WebSocketGraphQlTransportTests {
return session.send(session.receive()
.flatMap(webSocketMessage -> {
GraphQlWebSocketMessage message = this.codecDelegate.decode(webSocketMessage);
switch (message.resolvedType()) {
case CONNECTION_INIT:
return Flux.just(
GraphQlWebSocketMessage.connectionAck(null),
GraphQlWebSocketMessage.ping(null));
case SUBSCRIBE:
return Flux.just(GraphQlWebSocketMessage.next("1", this.response.toMap()));
case PONG:
return Flux.empty();
default:
return Flux.error(new IllegalStateException("Unexpected message: " + message));
}
return switch (message.resolvedType()) {
case CONNECTION_INIT -> Flux.just(
GraphQlWebSocketMessage.connectionAck(null),
GraphQlWebSocketMessage.ping(null));
case SUBSCRIBE -> Flux.just(GraphQlWebSocketMessage.next("1", this.response.toMap()));
case PONG -> Flux.empty();
default -> Flux.error(new IllegalStateException("Unexpected message: " + message));
};
})
.map(graphQlMessage -> this.codecDelegate.encode(session, graphQlMessage))
);