diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java b/spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java index b7216290..42591042 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java @@ -39,21 +39,26 @@ public interface WebSocketInterceptor extends WebInterceptor { * Handle the {@code "connection_init"} message at the start of a GraphQL over * WebSocket session and return an optional payload for the * {@code "connection_ack"} message to send back. + * @param sessionId the id of the WebSocket session * @param connectionInitPayload the payload from the {@code "connection_init"} message * @return the payload for the {@code "connection_ack"}, or empty */ - default Mono handleConnectionInitialization(Map connectionInitPayload) { + default Mono handleConnectionInitialization(String sessionId, Map connectionInitPayload) { return Mono.empty(); } /** - * Handle the {@code "complete"} message that clients send to stop listening - * to the subscription with the given id. - *

Note that the {@link org.reactivestreams.Publisher} for the subscription - * is automatically cancelled and there is no need to do that from here. + * Handle the {@code "complete"} message that a client sends to stop a + * subscription stream. The underlying {@link org.reactivestreams.Publisher} + * for the subscription is automatically cancelled. This callback is for any + * additional, or more centralized handling across subscriptions. + * @param sessionId the id of the WebSocket session + * @param subscriptionId the unique id for the subscription; correlates to the + * {@link WebInput#getId() requestId} from the original {@code "subscribe"} + * message that started the subscription * @return {@code Mono} for the completion of handling */ - default Mono handleCancelledSubscription(String subscriptionId) { + default Mono handleCancelledSubscription(String sessionId, String subscriptionId) { return Mono.empty(); } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java index 8ca2e77f..1313b812 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java @@ -141,14 +141,15 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { if (subscription != null) { subscription.cancel(); } - return this.webSocketInterceptor.handleCancelledSubscription(id).thenMany(Flux.empty()); + return this.webSocketInterceptor.handleCancelledSubscription(session.getId(), id) + .thenMany(Flux.empty()); } return Flux.empty(); case "connection_init": if (!connectionInitProcessed.compareAndSet(false, true)) { return GraphQlStatus.close(session, GraphQlStatus.TOO_MANY_INIT_REQUESTS_STATUS); } - return this.webSocketInterceptor.handleConnectionInitialization(payload) + return this.webSocketInterceptor.handleConnectionInitialization(session.getId(), payload) .defaultIfEmpty(Collections.emptyMap()) .map(ackPayload -> this.codecDelegate.encodeConnectionAck(session, ackPayload)) .flux() diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java index 45a000ad..e4919e9c 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java @@ -169,7 +169,8 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub if (subscription != null) { subscription.cancel(); } - this.webSocketInterceptor.handleCancelledSubscription(id).block(Duration.ofSeconds(10)); + this.webSocketInterceptor.handleCancelledSubscription(session.getId(), id) + .block(Duration.ofSeconds(10)); } return; case "connection_init": @@ -177,7 +178,7 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub GraphQlStatus.closeSession(session, GraphQlStatus.TOO_MANY_INIT_REQUESTS_STATUS); return; } - this.webSocketInterceptor.handleConnectionInitialization(payload) + this.webSocketInterceptor.handleConnectionInitialization(session.getId(), payload) .defaultIfEmpty(Collections.emptyMap()) .publishOn(sessionState.getScheduler()) // Serial blocking send via single thread .doOnNext(ackPayload -> { diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java b/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java index 54d0cc3e..02594e86 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java @@ -140,8 +140,8 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport { new WebSocketInterceptor() { @Override - public Mono handleConnectionInitialization(Map connectionInitPayload) { - Object value = connectionInitPayload.get("key"); + public Mono handleConnectionInitialization(String sessionId, Map payload) { + Object value = payload.get("key"); return Mono.just(Collections.singletonMap("key", value + " acknowledged")); } }); @@ -162,7 +162,7 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport { new WebSocketInterceptor() { @Override - public Mono handleConnectionInitialization(Map connectionInitPayload) { + public Mono handleConnectionInitialization(String sessionId, Map payload) { return Mono.error(new IllegalStateException()); } }); diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java b/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java index ecdc9d41..7d7141df 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java @@ -141,8 +141,8 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport { WebSocketInterceptor interceptor = new WebSocketInterceptor() { @Override - public Mono handleConnectionInitialization(Map connectionInitPayload) { - Object value = connectionInitPayload.get("key"); + public Mono handleConnectionInitialization(String sessionId, Map payload) { + Object value = payload.get("key"); return Mono.just(Collections.singletonMap("key", value + " acknowledged")); } }; @@ -166,7 +166,7 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport { WebSocketInterceptor interceptor = new WebSocketInterceptor() { @Override - public Mono handleConnectionInitialization(Map connectionInitPayload) { + public Mono handleConnectionInitialization(String sessionId, Map payload) { return Mono.error(new IllegalStateException()); } };