fix: update subs callback to SpringBoot 3.3

gradle.properties

@@ -3,12 +3,12 @@ version = 3.0-SNAPSHOT
 
 # dependencies
 annotationsVersion = 24.1.0
-graphQLJavaVersion = 22.0
+graphQLJavaVersion = 22.1
 mockWebServerVersion = 4.12.0
 protobufVersion = 4.26.1
 slf4jVersion = 2.0.13
-springBootVersion = 3.3.0-RC1
-springGraphQLVersion = 1.3.0-RC1
+springBootVersion = 3.3.0
+springGraphQLVersion = 1.3.0
 reactorVersion = 3.6.6
 
 # test dependencies

spring-subscription-callback/build.gradle.kts

@@ -4,13 +4,6 @@ plugins {
     id("com.apollographql.federation.java-conventions")
 }
 
-repositories {
-    mavenCentral()
-    maven {
-        url = uri("https://repo.spring.io/milestone")
-    }
-}
-
 val annotationsVersion: String by project
 val graphQLJavaVersion: String by project
 val mockWebServerVersion: String by project

spring-subscription-callback/src/main/java/com/apollographql/subscription/webflux/CallbackGraphQlHttpHandler.java

@@ -12,13 +12,27 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.jetbrains.annotations.NotNull;
+import org.reactivestreams.Publisher;
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.Decoder;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.graphql.ResponseError;
 import org.springframework.graphql.server.WebGraphQlHandler;
 import org.springframework.graphql.server.WebGraphQlRequest;
+import org.springframework.graphql.server.WebGraphQlResponse;
+import org.springframework.graphql.server.support.SerializableGraphQlRequest;
 import org.springframework.graphql.server.webflux.GraphQlHttpHandler;
+import org.springframework.graphql.support.DefaultExecutionGraphQlResponse;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
+import org.springframework.http.codec.CodecConfigurer;
+import org.springframework.http.codec.DecoderHttpMessageReader;
+import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
 import org.springframework.web.reactive.function.server.ServerRequest;
 import org.springframework.web.reactive.function.server.ServerResponse;
+import org.springframework.web.server.UnsupportedMediaTypeStatusException;
 import reactor.core.publisher.Mono;
 
 /**
@@ -47,22 +61,30 @@ public class CallbackGraphQlHttpHandler extends GraphQlHttpHandler {
 
   private final WebGraphQlHandler graphQlHandler;
   private final SubscriptionCallbackHandler subscriptionCallbackHandler;
+  @Nullable private final Decoder<?> decoder;
 
   public CallbackGraphQlHttpHandler(WebGraphQlHandler graphQlHandler) {
     this(graphQlHandler, new SubscriptionCallbackHandler(graphQlHandler));
   }
 
   public CallbackGraphQlHttpHandler(
       WebGraphQlHandler graphQlHandler, SubscriptionCallbackHandler subscriptionCallbackHandler) {
-    super(graphQlHandler);
+    this(graphQlHandler, subscriptionCallbackHandler, null);
+  }
+
+  public CallbackGraphQlHttpHandler(
+      WebGraphQlHandler graphQlHandler,
+      SubscriptionCallbackHandler subscriptionCallbackHandler,
+      @Nullable CodecConfigurer codecConfigurer) {
+    super(graphQlHandler, codecConfigurer);
     this.graphQlHandler = graphQlHandler;
     this.subscriptionCallbackHandler = subscriptionCallbackHandler;
+    this.decoder = (codecConfigurer != null) ? findJsonDecoder(codecConfigurer) : null;
   }
 
   @NotNull
   public Mono<ServerResponse> handleRequest(@NotNull ServerRequest serverRequest) {
-    return serverRequest
-        .bodyToMono(MAP_PARAMETERIZED_TYPE_REF)
+    return readRequest(serverRequest)
         .map(
             body ->
                 new WebGraphQlRequest(
@@ -81,9 +103,9 @@ public Mono<ServerResponse> handleRequest(@NotNull ServerRequest serverRequest)
               }
 
               // in order to correctly handle parsing of ANY requests (i.e. it is valid to define a
-              // document with query fragments first)
-              // we would need to parse it which is a much heavier operation, we may opt to do it in
-              // the future releases
+              // document with query fragments first) we would need to parse it which is a much
+              // heavier
+              // operation, we may opt to do it in the future releases
               if (graphQlRequest.getDocument().startsWith("subscription")) {
                 return parseSubscriptionCallbackExtension(graphQlRequest.getExtensions())
                     .flatMap(
@@ -96,14 +118,8 @@ public Mono<ServerResponse> handleRequest(@NotNull ServerRequest serverRequest)
                               .flatMap(
                                   success -> {
                                     if (success) {
-                                      var emptyResponse =
-                                          ExecutionResult.newExecutionResult().data(null).build();
-                                      var builder = ServerResponse.ok();
-                                      builder.header(
-                                          SUBSCRIPTION_PROTOCOL_HEADER,
-                                          SUBSCRIPTION_PROTOCOL_HEADER_VALUE);
-                                      builder.contentType(selectResponseMediaType(serverRequest));
-                                      return builder.bodyValue(emptyResponse.toSpecification());
+                                      return prepareResponse(
+                                          serverRequest, emptyResponse(graphQlRequest));
                                     } else {
                                       return ServerResponse.badRequest().build();
                                     }
@@ -121,25 +137,82 @@ public Mono<ServerResponse> handleRequest(@NotNull ServerRequest serverRequest)
                 return this.graphQlHandler
                     .handleRequest(graphQlRequest)
                     .flatMap(
-                        response -> {
+                        (response) -> {
                           if (logger.isDebugEnabled()) {
-                            logger.debug("Execution complete");
+                            List<ResponseError> errors = response.getErrors();
+                            logger.debug(
+                                "Execution result "
+                                    + (!CollectionUtils.isEmpty(errors)
+                                        ? "has errors: " + errors
+                                        : "is ready")
+                                    + ".");
                           }
-                          ServerResponse.BodyBuilder builder = ServerResponse.ok();
-                          builder.headers(headers -> headers.putAll(response.getResponseHeaders()));
-                          builder.contentType(selectResponseMediaType(serverRequest));
-                          return builder.bodyValue(response.toMap());
+
+                          return prepareResponse(serverRequest, response);
                         });
               }
             });
   }
 
-  private static MediaType selectResponseMediaType(ServerRequest serverRequest) {
-    for (MediaType accepted : serverRequest.headers().accept()) {
-      if (SUPPORTED_MEDIA_TYPES.contains(accepted)) {
-        return accepted;
-      }
+  private WebGraphQlResponse emptyResponse(WebGraphQlRequest request) {
+    var emptyResponse = ExecutionResult.newExecutionResult().data(null).build();
+    var emptyExecutionResponse =
+        new DefaultExecutionGraphQlResponse(request.toExecutionInput(), emptyResponse);
+    var emptyGraphQLResponse = new WebGraphQlResponse(emptyExecutionResponse);
+    emptyGraphQLResponse
+        .getResponseHeaders()
+        .add(SUBSCRIPTION_PROTOCOL_HEADER, SUBSCRIPTION_PROTOCOL_HEADER_VALUE);
+    return emptyGraphQLResponse;
+  }
+
+  // ========= CODE BELOW COPIED FROM SPRING =============
+  // Those are all private static methods so sadly we cannot access them directly :(
+  private static final ResolvableType REQUEST_TYPE =
+      ResolvableType.forClass(SerializableGraphQlRequest.class);
+
+  // copy from protected HttpCodecDelegate class
+  private static Decoder<?> findJsonDecoder(CodecConfigurer configurer) {
+    return configurer.getReaders().stream()
+        .filter((reader) -> reader.canRead(REQUEST_TYPE, MediaType.APPLICATION_JSON))
+        .map((reader) -> ((DecoderHttpMessageReader<?>) reader).getDecoder())
+        .findFirst()
+        .orElseThrow(() -> new IllegalArgumentException("No JSON Decoder"));
+  }
+
+  @SuppressWarnings("unchecked")
+  Mono<SerializableGraphQlRequest> decode(
+      Publisher<DataBuffer> inputStream, MediaType contentType) {
+    return (Mono<SerializableGraphQlRequest>)
+        this.decoder.decodeToMono(inputStream, REQUEST_TYPE, contentType, null);
+  }
+
+  // copy from AbstractGraphQlHttpHandler
+  private Mono<SerializableGraphQlRequest> readRequest(ServerRequest serverRequest) {
+    if (this.decoder != null) {
+      MediaType contentType =
+          serverRequest.headers().contentType().orElse(MediaType.APPLICATION_JSON);
+      return decode(serverRequest.bodyToFlux(DataBuffer.class), contentType);
+    } else {
+      return serverRequest
+          .bodyToMono(SerializableGraphQlRequest.class)
+          .onErrorResume(
+              UnsupportedMediaTypeStatusException.class,
+              (ex) -> applyApplicationGraphQlFallback(ex, serverRequest));
     }
-    return MediaType.APPLICATION_JSON;
+  }
+
+  // copy from AbstractGraphQlHttpHandler
+  private static Mono<SerializableGraphQlRequest> applyApplicationGraphQlFallback(
+      UnsupportedMediaTypeStatusException ex, ServerRequest request) {
+
+    // Spec requires application/json but some clients still use application/graphql
+    return "application/graphql".equals(request.headers().firstHeader(HttpHeaders.CONTENT_TYPE))
+        ? ServerRequest.from(request)
+            .headers((headers) -> headers.setContentType(MediaType.APPLICATION_JSON))
+            .body(request.bodyToFlux(DataBuffer.class))
+            .build()
+            .bodyToMono(SerializableGraphQlRequest.class)
+            .log()
+        : Mono.error(ex);
   }
 }