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

Add sessionId to WebSocketInterceptor callbacks

Browse Source 
See gh-276
This commit is contained in:
rstoyanchev
2022-03-08 07:29:10 +00:00
parent c6d10d3ab8
commit 35359b0b24
5 changed files with 23 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java
View File

@@ -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<Object> handleConnectionInitialization(Map<String, Object> connectionInitPayload) {
default Mono<Object> handleConnectionInitialization(String sessionId, Map<String, Object> connectionInitPayload) {
return Mono.empty();
}
/**
* Handle the {@code "complete"} message that clients send to stop listening
* to the subscription with the given id.
* <p>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<Void> handleCancelledSubscription(String subscriptionId) {
default Mono<Void> handleCancelledSubscription(String sessionId, String subscriptionId) {
return Mono.empty();
}

5
spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java
View File

@@ -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()

5
spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java
View File

@@ -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 -> {

6
spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java
View File

@@ -140,8 +140,8 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport {
new WebSocketInterceptor() {
@Override
public Mono<Object> handleConnectionInitialization(Map<String, Object> connectionInitPayload) {
Object value = connectionInitPayload.get("key");
public Mono<Object> handleConnectionInitialization(String sessionId, Map<String, Object> 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<Object> handleConnectionInitialization(Map<String, Object> connectionInitPayload) {
public Mono<Object> handleConnectionInitialization(String sessionId, Map<String, Object> payload) {
return Mono.error(new IllegalStateException());
}
});

6
spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java
View File

@@ -141,8 +141,8 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport {
WebSocketInterceptor interceptor = new WebSocketInterceptor() {
@Override
public Mono<Object> handleConnectionInitialization(Map<String, Object> connectionInitPayload) {
Object value = connectionInitPayload.get("key");
public Mono<Object> handleConnectionInitialization(String sessionId, Map<String, Object> 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<Object> handleConnectionInitialization(Map<String, Object> connectionInitPayload) {
public Mono<Object> handleConnectionInitialization(String sessionId, Map<String, Object> payload) {
return Mono.error(new IllegalStateException());
}
};