Add Server-Sent Events transport

Prior to this commit, the WebFlux and WebMVC infrastructure would only
support subscriptions over the WebSocket and RSocket transports.

This commit adds the `GraphQlSseHandler` implementations for both web
frameworks. This handler will send GraphQL responses as as stream of
Server-Sent Events, over an HTTP response with the "text/event-stream"
content type.

This implementation only supports the "Distinct connections mode" and
will reject all operations other than Subscriptions.

This commit also enhances the `HttpGraphQlTransport` client transport to
support subscriptions over this new protocol.

Closes gh-309

Author: bclozel

build.gradle

@@ -2,7 +2,7 @@ description = "Spring for GraphQL"
 
 ext {
 	moduleProjects = [project(":spring-graphql"), project(":spring-graphql-test")]
-	springFrameworkVersion = "6.1.3"
+	springFrameworkVersion = "6.1.4-SNAPSHOT"
 	graphQlJavaVersion = "21.3"
 	springBootVersion = "3.2.2"
 }

platform/build.gradle

@@ -18,7 +18,7 @@ dependencies {
 	api(platform("io.rsocket:rsocket-bom:1.1.4"))
 	api(platform("org.jetbrains.kotlin:kotlin-bom:${kotlinVersion}"))
 	api(platform("org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.7.3"))
-	api(platform("org.junit:junit-bom:5.10.1"))
+	api(platform("org.junit:junit-bom:5.10.2"))
 	api(platform("org.mockito:mockito-bom:5.8.0"))
 	api(platform("org.testcontainers:testcontainers-bom:1.19.4"))
 	api(platform("org.apache.logging.log4j:log4j-bom:2.22.1"))
@@ -39,6 +39,8 @@ dependencies {
 		api("org.assertj:assertj-core:3.24.2")
 		api("com.jayway.jsonpath:json-path:2.8.0")
 		api("org.skyscreamer:jsonassert:1.5.1")
+		api("org.awaitility:awaitility:4.2.0")
+		api("com.squareup.okhttp3:mockwebserver:4.12.0")
 
 		api("com.h2database:h2:2.1.214")
 		api("org.hibernate:hibernate-core:6.4.2.Final")

spring-graphql-docs/modules/ROOT/pages/client.adoc

@@ -467,10 +467,12 @@ You can use the `GraphQlClient` xref:client.adoc#client.graphqlclient.builder[Bu
 [[client.subscriptions]]
 == Subscription Requests
 
-`GraphQlClient` can execute subscriptions over transports that support it. Only
-the WebSocket and RSocket transports support GraphQL subscriptions, so you'll need to
-create a xref:client.adoc#client.websocketgraphqlclient[WebSocketGraphQlClient] or
-xref:client.adoc#client.rsocketgraphqlclient[RSocketGraphQlClient].
+Subscription requests require a client transport that is capable of streaming data.
+You will need to create a `GraphQlClient` that support this:
+
+- xref:client.adoc#client.httpgraphqlclient[HttpGraphQlClient] with Server-Sent Events
+- xref:client.adoc#client.websocketgraphqlclient[WebSocketGraphQlClient] with WebSocket
+- xref:client.adoc#client.rsocketgraphqlclient[RSocketGraphQlClient] with RSocket
 
 
 

spring-graphql-docs/modules/ROOT/pages/transports.adoc

@@ -32,6 +32,27 @@ it contains, for the actual config.
 The 1.0.x branch of this repository contains a Spring MVC
 {github-10x-branch}/samples/webmvc-http[HTTP sample] application.
 
+
+[[server.transports.sse]]
+== Server-Sent Events
+
+`GraphQlSseHandler` is very similar to the HTTP handler listed above, but this time handling GraphQL requests over HTTP
+using the Server-Sent Events protocol. With this transport, clients must send HTTP POST requests to the endpoint with
+`"application/json"` as content type and GraphQL request details included as JSON in the request body; the only
+difference with the vanilla HTTP variant is that the client must send `"text/event-stream"` as the `"Accept"` request
+header. The response will be sent as one or more Server-Sent Event(s).
+
+This is also defined in the proposed
+https://github.com/graphql/graphql-over-http/blob/main/rfcs/GraphQLOverSSE.md[GraphQL over HTTP] specification.
+Spring for GraphQL only implements the "Distinct connections mode", so applications must consider scalability concerns
+and whether adopting HTTP/2 as the underlying transport would help.
+
+The main use case for `GraphQlSseHandler` is an alternative to the
+xref:transports.adoc#server.transports.websocket[WebSocket transport], receiving a stream of items as a response to a
+subscription operation. Other types of operations, like queries and mutations, are not supported here and should be
+using the plain JSON over HTTP transport variant.
+
+
 [[server.transports.http.fileupload]]
 === File Upload
 

spring-graphql-test/build.gradle

@@ -32,7 +32,6 @@ dependencies {
 	testImplementation 'io.projectreactor.netty:reactor-netty'
 	testImplementation 'io.rsocket:rsocket-transport-local'
 	testImplementation 'io.micrometer:context-propagation'
-	testImplementation 'com.squareup.okhttp3:mockwebserver:3.14.9'
 	testImplementation 'com.fasterxml.jackson.core:jackson-databind'
 
 	testRuntimeOnly 'org.apache.logging.log4j:log4j-core'

spring-graphql/build.gradle

@@ -37,6 +37,7 @@ dependencies {
 	testImplementation 'org.junit.jupiter:junit-jupiter'
 	testImplementation 'org.assertj:assertj-core'
 	testImplementation 'org.mockito:mockito-core'
+	testImplementation 'org.awaitility:awaitility'
 	testImplementation 'io.projectreactor:reactor-test'
 	testImplementation 'org.springframework:spring-core-test'
 	testImplementation 'org.springframework:spring-messaging'
@@ -63,20 +64,22 @@ dependencies {
 	testImplementation 'com.querydsl:querydsl-core'
 	testImplementation 'com.querydsl:querydsl-collections'
 	testImplementation 'jakarta.servlet:jakarta.servlet-api'
-	testImplementation 'com.squareup.okhttp3:mockwebserver:3.14.9'
+	testImplementation 'com.squareup.okhttp3:mockwebserver'
 	testImplementation 'io.rsocket:rsocket-transport-local'
 	testImplementation 'jakarta.persistence:jakarta.persistence-api'
 	testImplementation 'jakarta.validation:jakarta.validation-api'
 	testImplementation 'com.jayway.jsonpath:json-path'
 	testImplementation 'com.fasterxml.jackson.core:jackson-databind'
-	testImplementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310")
+	testImplementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
 	testImplementation 'org.apache.tomcat.embed:tomcat-embed-el:10.0.21'
 	testImplementation 'com.apollographql.federation:federation-graphql-java-support'
 
 	testRuntimeOnly 'org.apache.logging.log4j:log4j-core'
 	testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl'
 
 	testFixturesApi 'org.springframework:spring-webflux'
+	testFixturesApi 'org.junit.jupiter:junit-jupiter-engine'
+	testFixturesApi 'com.squareup.okhttp3:mockwebserver'
 	testFixturesApi 'com.fasterxml.jackson.core:jackson-databind'
 }
 

spring-graphql/src/main/java/org/springframework/graphql/client/HttpGraphQlTransport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -26,6 +26,7 @@
 import org.springframework.graphql.GraphQlResponse;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.util.Assert;
 import org.springframework.web.reactive.function.client.WebClient;
 
@@ -37,13 +38,17 @@
  * see {@link WebSocketGraphQlTransport} and {@link RSocketGraphQlTransport}.
  *
  * @author Rossen Stoyanchev
+ * @author Brian Clozel
  * @since 1.0.0
  */
 final class HttpGraphQlTransport implements GraphQlTransport {
 
 	private static final ParameterizedTypeReference<Map<String, Object>> MAP_TYPE =
 			new ParameterizedTypeReference<Map<String, Object>>() {};
 
+	private static final ParameterizedTypeReference<ServerSentEvent<Map<String, Object>>> SSE_TYPE =
+			new ParameterizedTypeReference<ServerSentEvent<Map<String, Object>>>() {};
+
 	// To be removed in favor of Framework's MediaType.APPLICATION_GRAPHQL_RESPONSE
 	private static final MediaType APPLICATION_GRAPHQL_RESPONSE =
 			new MediaType("application", "graphql-response+json");
@@ -87,7 +92,19 @@ public Mono<GraphQlResponse> execute(GraphQlRequest request) {
 
 	@Override
 	public Flux<GraphQlResponse> executeSubscription(GraphQlRequest request) {
-		throw new UnsupportedOperationException("Subscriptions not supported over HTTP");
+		return this.webClient.post()
+				.contentType(this.contentType)
+				.accept(MediaType.TEXT_EVENT_STREAM)
+				.bodyValue(request.toMap())
+				.attributes(attributes -> {
+					if (request instanceof ClientGraphQlRequest clientRequest) {
+						attributes.putAll(clientRequest.getAttributes());
+					}
+				})
+				.retrieve()
+				.bodyToFlux(SSE_TYPE)
+				.takeWhile(event -> "next".equals(event.event()))
+				.map(event -> new ResponseMapGraphQlResponse(event.data()));
 	}
 
 }

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

@@ -18,13 +18,11 @@
 
 import java.util.Arrays;
 import java.util.List;
-import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Mono;
 
-import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.graphql.server.WebGraphQlHandler;
 import org.springframework.graphql.server.WebGraphQlRequest;
 import org.springframework.graphql.server.support.SerializableGraphQlRequest;
@@ -44,16 +42,9 @@ public class GraphQlHttpHandler {
 
 	private static final Log logger = LogFactory.getLog(GraphQlHttpHandler.class);
 
-	// To be removed in favor of Framework's MediaType.APPLICATION_GRAPHQL_RESPONSE
-	private static final MediaType APPLICATION_GRAPHQL_RESPONSE =
-			new MediaType("application", "graphql-response+json");
-
-	private static final ParameterizedTypeReference<Map<String, Object>> MAP_PARAMETERIZED_TYPE_REF =
-			new ParameterizedTypeReference<Map<String, Object>>() {};
-
 	@SuppressWarnings("removal")
 	private static final List<MediaType> SUPPORTED_MEDIA_TYPES =
-			Arrays.asList(APPLICATION_GRAPHQL_RESPONSE, MediaType.APPLICATION_JSON, MediaType.APPLICATION_GRAPHQL);
+			Arrays.asList(MediaType.APPLICATION_GRAPHQL_RESPONSE, MediaType.APPLICATION_JSON, MediaType.APPLICATION_GRAPHQL);
 
 	private final WebGraphQlHandler graphQlHandler;
 

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

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2020-2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.server.webflux;
+
+
+import java.util.Collections;
+import java.util.Map;
+
+import graphql.ErrorType;
+import graphql.ExecutionResult;
+import graphql.GraphQLError;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.graphql.execution.SubscriptionPublisherException;
+import org.springframework.graphql.server.WebGraphQlHandler;
+import org.springframework.graphql.server.WebGraphQlRequest;
+import org.springframework.graphql.server.support.SerializableGraphQlRequest;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+
+/**
+ * GraphQL handler that supports the
+ * <a href="https://github.com/graphql/graphql-over-http/blob/main/rfcs/GraphQLOverSSE.md">GraphQL
+ * Server-Sent Events Protocol</a> and to be exposed as a WebFlux.fn endpoint via
+ * {@link org.springframework.web.reactive.function.server.RouterFunctions}.
+ *
+ * @author Brian Clozel
+ * @since 1.3.0
+ */
+public class GraphQlSseHandler {
+
+    private static final Log logger = LogFactory.getLog(GraphQlSseHandler.class);
+
+    private static final Mono<ServerSentEvent<Map<String, Object>>> COMPLETE_EVENT = Mono.just(ServerSentEvent.<Map<String, Object>>builder().event("complete").build());
+
+    private final WebGraphQlHandler graphQlHandler;
+
+
+    public GraphQlSseHandler(WebGraphQlHandler graphQlHandler) {
+        Assert.notNull(graphQlHandler, "WebGraphQlHandler is required");
+        this.graphQlHandler = graphQlHandler;
+    }
+
+    /**
+     * Handle GraphQL requests over HTTP using the Server-Sent Events protocol.
+     *
+     * @param serverRequest the incoming HTTP request
+     * @return the HTTP response
+     */
+    @SuppressWarnings("unchecked")
+    public Mono<ServerResponse> handleRequest(ServerRequest serverRequest) {
+        Flux<ServerSentEvent<Map<String, Object>>> data = serverRequest.bodyToMono(SerializableGraphQlRequest.class)
+                .flatMap(body -> {
+                    WebGraphQlRequest graphQlRequest = new WebGraphQlRequest(
+                            serverRequest.uri(), serverRequest.headers().asHttpHeaders(),
+                            serverRequest.cookies(), serverRequest.attributes(), body,
+                            serverRequest.exchange().getRequest().getId(),
+                            serverRequest.exchange().getLocaleContext().getLocale());
+                    if (logger.isDebugEnabled()) {
+                        logger.debug("Executing: " + graphQlRequest);
+                    }
+                    return this.graphQlHandler.handleRequest(graphQlRequest);
+                })
+                .flatMapMany(response -> {
+                    if (logger.isDebugEnabled()) {
+                        logger.debug("Execution result ready"
+                                + (!CollectionUtils.isEmpty(response.getErrors()) ? " with errors: " + response.getErrors() : "")
+                                + ".");
+                    }
+                    if (response.getData() instanceof Publisher) {
+                        // Subscription
+                        return Flux.from((Publisher<ExecutionResult>) response.getData()).map(ExecutionResult::toSpecification);
+                    }
+                    if (logger.isDebugEnabled()) {
+                        logger.debug("Only subscriptions are supported, DataFetcher must return a Publisher type");
+                    }
+                    // Single response (query or mutation) are not supported
+                    String errorMessage = "SSE transport only supports Subscription operations";
+                    GraphQLError unsupportedOperationError = GraphQLError.newError().errorType(ErrorType.OperationNotSupported)
+                            .message(errorMessage).build();
+                    return Flux.error(new SubscriptionPublisherException(Collections.singletonList(unsupportedOperationError),
+                            new IllegalArgumentException(errorMessage)));
+                })
+                .onErrorResume(SubscriptionPublisherException.class, exc -> {
+                    ExecutionResult errorResult = ExecutionResult.newExecutionResult().errors(exc.getErrors()).build();
+                    return Flux.just(errorResult.toSpecification());
+                })
+                .map(event -> ServerSentEvent.builder(event).event("next").build());
+
+        Flux<ServerSentEvent<Map<String, Object>>> body = data.concatWith(COMPLETE_EVENT);
+        return ServerResponse.ok().contentType(MediaType.TEXT_EVENT_STREAM).body(BodyInserters.fromServerSentEvents(body))
+                .onErrorResume(Throwable.class, exc -> ServerResponse.badRequest().build());
+    }
+
+}

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

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2020-2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.server.webmvc;
+
+import java.io.IOException;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.Cookie;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.graphql.GraphQlRequest;
+import org.springframework.graphql.server.WebGraphQlHandler;
+import org.springframework.graphql.server.support.SerializableGraphQlRequest;
+import org.springframework.http.HttpCookie;
+import org.springframework.util.AlternativeJdkIdGenerator;
+import org.springframework.util.Assert;
+import org.springframework.util.IdGenerator;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.server.ServerWebInputException;
+import org.springframework.web.servlet.function.ServerRequest;
+
+/**
+ * Abstract class for GraphQL Handler implementations using the HTTP transport.
+ *
+ * @author Brian Clozel
+ * @since 1.3.0
+ */
+abstract class AbstractGraphQlHttpHandler {
+
+    protected final Log logger = LogFactory.getLog(getClass());
+
+    protected final IdGenerator idGenerator = new AlternativeJdkIdGenerator();
+
+    protected final WebGraphQlHandler graphQlHandler;
+
+
+    AbstractGraphQlHttpHandler(WebGraphQlHandler graphQlHandler) {
+        Assert.notNull(graphQlHandler, "WebGraphQlHandler is required");
+        this.graphQlHandler = graphQlHandler;
+    }
+
+    protected static MultiValueMap<String, HttpCookie> initCookies(ServerRequest serverRequest) {
+        MultiValueMap<String, Cookie> source = serverRequest.cookies();
+        MultiValueMap<String, HttpCookie> target = new LinkedMultiValueMap<>(source.size());
+        source.values().forEach(cookieList -> cookieList.forEach(cookie -> {
+            HttpCookie httpCookie = new HttpCookie(cookie.getName(), cookie.getValue());
+            target.add(cookie.getName(), httpCookie);
+        }));
+        return target;
+    }
+
+    protected static GraphQlRequest readBody(ServerRequest request) throws ServletException {
+        try {
+            return request.body(SerializableGraphQlRequest.class);
+        }
+        catch (IOException ex) {
+            throw new ServerWebInputException("I/O error while reading request body", null, ex);
+        }
+    }
+
+}