From 3b1d1abfb9cd695649174861671058017ac1127e Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Tue, 18 Apr 2023 09:30:34 +0100 Subject: [PATCH] Polishing in WebSocket support --- .../DefaultWebSocketGraphQlClientBuilder.java | 4 +- .../WebSocketGraphQlClientInterceptor.java | 6 +- .../client/WebSocketGraphQlTransport.java | 65 +++++-------------- .../server/WebSocketGraphQlInterceptor.java | 14 ++-- .../webflux/GraphQlWebSocketHandler.java | 16 +++-- .../webmvc/GraphQlWebSocketHandler.java | 24 +++---- .../client/MockGraphQlWebSocketServer.java | 14 ++-- .../WebSocketGraphQlTransportTests.java | 22 +++---- 8 files changed, 68 insertions(+), 97 deletions(-) diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java index fc519cac..90204385 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java @@ -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 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"); diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClientInterceptor.java b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClientInterceptor.java index 9e3c28a1..f5228d4e 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClientInterceptor.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClientInterceptor.java @@ -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. * + *

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 */ 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 f89a0041..f6e1067a 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 @@ -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 initGraphQlSession( URI uri, HttpHeaders headers, WebSocketClient client, GraphQlSessionHandler handler) { @@ -219,6 +220,7 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { } + @SuppressWarnings({"ReactorZipWithMonoVoid", "ReactiveStreamsThrowInOperator"}) @Override public Mono 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 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 responseSink; - - SingleResponseRequestState(GraphQlRequest request, MonoSink responseSink) { - this.request = request; - this.responseSink = responseSink; - } - - @Override - public GraphQlRequest getRequest() { - return this.request; - } + private record SingleResponseRequestState( + GraphQlRequest request, MonoSink 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 responseSink; - - SubscriptionRequestState(GraphQlRequest request, FluxSink responseSink) { - this.request = request; - this.responseSink = responseSink; - } - - @Override - public GraphQlRequest getRequest() { - return request; - } + private record SubscriptionRequestState( + GraphQlRequest request, FluxSink responseSink) implements RequestState { @Override public void handleResponse(GraphQlResponse response) { diff --git a/spring-graphql/src/main/java/org/springframework/graphql/server/WebSocketGraphQlInterceptor.java b/spring-graphql/src/main/java/org/springframework/graphql/server/WebSocketGraphQlInterceptor.java index 1c3107ec..def46297 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/server/WebSocketGraphQlInterceptor.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/server/WebSocketGraphQlInterceptor.java @@ -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. + * + *

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 handleCancelledSubscription(WebSocketSessionInfo sessionInfo, String subscriptionId) { diff --git a/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/GraphQlWebSocketHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/GraphQlWebSocketHandler.java index f6151fa7..0f9a40ac 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/GraphQlWebSocketHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/GraphQlWebSocketHandler.java @@ -101,6 +101,7 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { } + @SuppressWarnings("CallingSubscribeInNonBlockingScope") @Override public Mono handle(WebSocketSession session) { HandshakeInfo handshakeInfo = session.getHandshakeInfo(); @@ -140,7 +141,7 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { String id = message.getId(); Map 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); + } } })); } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlWebSocketHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlWebSocketHandler.java index 42b2eba3..c4184cb7 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlWebSocketHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlWebSocketHandler.java @@ -191,7 +191,7 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub Map 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 diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java b/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java index 712a3b3b..26b8159c 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java @@ -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 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 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)); + } } } diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java b/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java index bd6371bc..2b99674f 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java @@ -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)) );