From 0d672c1383aff5874619c68c28f3848cd6b8f47c Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Mon, 17 Mar 2025 18:13:54 +0000 Subject: [PATCH] SSE handlers support keep-alive Closes gh-1048 --- .../webflux/GraphQlWebSocketHandler.java | 4 +- .../server/webmvc/GraphQlSseHandler.java | 40 +++++++++++-------- .../webmvc/GraphQlWebSocketHandler.java | 4 +- .../webflux/GraphQlSseHandlerTests.java | 7 ++-- 4 files changed, 31 insertions(+), 24 deletions(-) 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 6771536c..264ca9f1 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 @@ -104,8 +104,8 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { * @param codecConfigurer codec configurer for JSON encoding and decoding * @param connectionInitTimeout how long to wait after the establishment of * the WebSocket for the {@code "connection_ini"} message from the client. - * @param keepAliveDuration how frequently to send ping messages; if not - * set then ping messages are not sent. + * @param keepAliveDuration how frequently to send ping messages when no + * other messages are sent * @since 1.3 */ public GraphQlWebSocketHandler( diff --git a/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlSseHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlSseHandler.java index fde91864..8171b983 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlSseHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlSseHandler.java @@ -26,6 +26,7 @@ import graphql.ErrorType; import graphql.ExecutionResult; import graphql.GraphQLError; import graphql.GraphqlErrorBuilder; +import org.apache.commons.logging.Log; import org.reactivestreams.Publisher; import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; @@ -125,8 +126,8 @@ public class GraphQlSseHandler extends AbstractGraphQlHttpHandler { }); return ((this.timeout != null) ? - ServerResponse.sse(SseSubscriber.connect(resultFlux, this.keepAliveDuration), this.timeout) : - ServerResponse.sse(SseSubscriber.connect(resultFlux, this.keepAliveDuration))); + ServerResponse.sse(SseSubscriber.connect(resultFlux, this.logger, this.keepAliveDuration), this.timeout) : + ServerResponse.sse(SseSubscriber.connect(resultFlux, this.logger, this.keepAliveDuration))); } @@ -137,9 +138,12 @@ public class GraphQlSseHandler extends AbstractGraphQlHttpHandler { private final ServerResponse.SseBuilder sseBuilder; - private SseSubscriber(ServerResponse.SseBuilder sseBuilder) { + private final Log logger; + + private SseSubscriber(ServerResponse.SseBuilder sseBuilder, Log logger) { this.sseBuilder = sseBuilder; this.sseBuilder.onTimeout(() -> cancelWithError(new AsyncRequestTimeoutException())); + this.logger = logger; } @Override @@ -180,20 +184,24 @@ public class GraphQlSseHandler extends AbstractGraphQlHttpHandler { @Override protected void hookOnError(Throwable ex) { - sendNext(exceptionToResultMap(ex)); + Map errorMap; + if (ex instanceof SubscriptionPublisherException spe) { + errorMap = spe.toMap(); + } + else { + if (this.logger.isErrorEnabled()) { + this.logger.error("Unresolved " + ex.getClass().getSimpleName(), ex); + } + errorMap = GraphqlErrorBuilder.newError() + .message("Subscription error") + .errorType(org.springframework.graphql.execution.ErrorType.INTERNAL_ERROR) + .build() + .toSpecification(); + } + sendNext(errorMap); sendComplete(); } - private static Map exceptionToResultMap(Throwable ex) { - return ((ex instanceof SubscriptionPublisherException spe) ? - spe.toMap() : - GraphqlErrorBuilder.newError() - .message("Subscription error") - .errorType(org.springframework.graphql.execution.ErrorType.INTERNAL_ERROR) - .build() - .toSpecification()); - } - private void sendComplete() { try { this.sseBuilder.event("complete").data(""); @@ -210,10 +218,10 @@ public class GraphQlSseHandler extends AbstractGraphQlHttpHandler { } static Consumer connect( - Flux> resultFlux, @Nullable Duration keepAliveDuration) { + Flux> resultFlux, Log logger, @Nullable Duration keepAliveDuration) { return (sseBuilder) -> { - SseSubscriber subscriber = new SseSubscriber(sseBuilder); + SseSubscriber subscriber = new SseSubscriber(sseBuilder, logger); if (keepAliveDuration != null) { KeepAliveHandler handler = new KeepAliveHandler(keepAliveDuration); handler.compose(resultFlux).subscribe(subscriber); 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 f4109c1e..43ef24c4 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 @@ -129,8 +129,8 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub * @param converter for JSON encoding and decoding * @param connectionInitTimeout how long to wait after the establishment of * the WebSocket for the {@code "connection_ini"} message from the client. - * @param keepAliveDuration how frequently to send ping messages; if not - * set then ping messages are not sent. + * @param keepAliveDuration how frequently to send ping messages when no + * other messages are sent * @since 1.3 */ public GraphQlWebSocketHandler( diff --git a/spring-graphql/src/test/java/org/springframework/graphql/server/webflux/GraphQlSseHandlerTests.java b/spring-graphql/src/test/java/org/springframework/graphql/server/webflux/GraphQlSseHandlerTests.java index d7707e13..680127d2 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/server/webflux/GraphQlSseHandlerTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/server/webflux/GraphQlSseHandlerTests.java @@ -41,7 +41,6 @@ import org.springframework.mock.web.reactive.function.server.MockServerRequest; import org.springframework.mock.web.server.MockServerWebExchange; import org.springframework.web.reactive.function.server.ServerResponse; import org.springframework.web.reactive.result.view.ViewResolver; -import org.springframework.web.server.ServerWebExchange; import static org.assertj.core.api.Assertions.assertThat; @@ -181,9 +180,9 @@ class GraphQlSseHandlerTests { MockServerRequest serverRequest = MockServerRequest.builder() .exchange(exchange) - .uri(((ServerWebExchange) exchange).getRequest().getURI()) - .method(((ServerWebExchange) exchange).getRequest().getMethod()) - .headers(((ServerWebExchange) exchange).getRequest().getHeaders()) + .uri(exchange.getRequest().getURI()) + .method(exchange.getRequest().getMethod()) + .headers(exchange.getRequest().getHeaders()) .body(Mono.just(body)); handler.handleRequest(serverRequest)