SSE handlers support keep-alive

Closes gh-1048

Author: rstoyanchev

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

@@ -104,8 +104,8 @@ public GraphQlWebSocketHandler(
 	 * @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(

spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlSseHandler.java

@@ -26,6 +26,7 @@
 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 @@ protected ServerResponse prepareResponse(
 		});
 
 		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 @@ private static final class SseSubscriber extends BaseSubscriber<Map<String, Obje
 
 		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 @@ private void cancelWithError(Throwable ex) {
 
 		@Override
 		protected void hookOnError(Throwable ex) {
-			sendNext(exceptionToResultMap(ex));
+			Map<String, Object> 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<String, Object> 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 @@ protected void hookOnComplete() {
 		}
 
 		static Consumer<ServerResponse.SseBuilder> connect(
-				Flux<Map<String, Object>> resultFlux, @Nullable Duration keepAliveDuration) {
+				Flux<Map<String, Object>> 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);

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

@@ -129,8 +129,8 @@ public GraphQlWebSocketHandler(
 	 * @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(

spring-graphql/src/test/java/org/springframework/graphql/server/webflux/GraphQlSseHandlerTests.java

@@ -41,7 +41,6 @@
 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 @@ private MockServerHttpResponse handleRequest(
 
 		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)