feat: add the GraphQL WebSocket server (G4, graphql-transport-ws)

changes.md

@@ -92,7 +92,11 @@ Changes log
           result requested at a time, client disconnects cancel the subscription, and finite
           limits apply (64 active subscriptions and 1 MiB of queued bytes per subscription by
           default). `GraphqlClientAdapter.subscribe` returns a `Publisher<ExecutionResult>`
-          honoring demand; cancelling it aborts the request. Issue #1581.
+          honoring demand; cancelling it aborts the request. With `GraphqlTransport.WEBSOCKET`,
+          the server adapter also accepts `graphql-transport-ws` WebSocket upgrades on its own path
+          (HTTP/1.1 only), with several operations per connection, a `connection_init` timeout
+          and the protocol's close codes; `org.restlet.ext.websocket` is an optional dependency,
+          loaded only then. Issue #1581.
         - Added the `NettyConnectionUpgrade` engine hook: a Restlet returning a "101 Switching
           Protocols" response with this response attribute takes over the HTTP/1.1 connection
           (`NettyUpgradeContext`), which the WebSocket extension relies on.

org.restlet.ext.graphql/pom.xml

@@ -29,11 +29,46 @@
 			<artifactId>jackson-databind</artifactId>
 			<version>${lib-jackson3-version}</version>
 		</dependency>
+		<dependency>
+			<groupId>org.restlet</groupId>
+			<artifactId>org.restlet.ext.websocket</artifactId>
+			<version>${project.version}</version>
+			<!-- Only needed for GraphqlTransport.WEBSOCKET -->
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.junit.jupiter</groupId>
 			<artifactId>junit-jupiter-api</artifactId>
 			<version>${lib-junit-version}</version>
 			<scope>test</scope>
 		</dependency>
 	</dependencies>
+	<profiles>
+		<profile>
+			<id>m2e-test-modules</id>
+			<activation>
+				<property><name>m2e.version</name></property>
+			</activation>
+			<build>
+				<plugins>
+					<plugin>
+						<artifactId>maven-compiler-plugin</artifactId>
+						<executions>
+							<execution>
+								<id>default-testCompile</id>
+								<configuration>
+									<!-- Keep the JDK HTTP client and server test-only in m2e/JDT;
+									     Maven already compiles tests on the classpath. -->
+									<compilerArgs combine.children="append">
+										<arg>--add-reads</arg>
+										<arg>org.restlet.ext.graphql=java.net.http,jdk.httpserver</arg>
+									</compilerArgs>
+								</configuration>
+							</execution>
+						</executions>
+					</plugin>
+				</plugins>
+			</build>
+		</profile>
+	</profiles>
 </project>

org.restlet.ext.graphql/src/main/java/module-info.java

@@ -16,6 +16,9 @@
     requires transitive com.graphqljava;
     requires transitive org.reactivestreams;
 
+    // Optional: only loaded when GraphqlTransport.WEBSOCKET is enabled.
+    requires static org.restlet.ext.websocket;
+
     // JSON is an implementation detail: required, not re-exported.
     requires tools.jackson.core;
     requires tools.jackson.databind;

org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java

@@ -83,6 +83,12 @@
  * {@code complete} event. A client disconnect, detected on write or by heartbeats, cancels the
  * subscription, and stopping the adapter ends its open streams.
  *
+ * <p>With {@link GraphqlTransport#WEBSOCKET}, the adapter also accepts WebSocket upgrades offering
+ * the {@code graphql-transport-ws} subprotocol on its own path, on HTTP/1.1 connections. The
+ * upgrade request goes through the same filters and authenticators, and the payload of the {@code
+ * connection_init} message is available under {@link RestletGraphql#CONNECTION_INIT_PAYLOAD}.
+ * Subscriptions share the limits of SSE ones; stopping the adapter closes the connections.
+ *
  * <p>Operations run on the thread of the call, a virtual thread by default. Asynchronous data
  * fetchers are joined on that thread. The {@code graphql.GraphQLContext} of each operation holds
  * the Restlet request under {@link RestletGraphql#REQUEST} and its client info under {@link
@@ -104,6 +110,9 @@
     /** The default maximum number of bytes queued for a subscription: 1 MiB. */
     public static final long DEFAULT_MAX_QUEUED_BYTES = 1L << 20;
 
+    /** The default time allowed to WebSocket clients to initialize their connection: 3 seconds. */
+    public static final Duration DEFAULT_CONNECTION_INIT_TIMEOUT = Duration.ofSeconds(3);
+
     /** The default interval of event stream heartbeats: 15 seconds. */
     public static final Duration DEFAULT_HEARTBEAT_INTERVAL = Duration.ofSeconds(15);
 
@@ -130,6 +139,10 @@
 
     private final Set<GraphqlEventStream> openStreams = ConcurrentHashMap.newKeySet();
 
+    private volatile Duration connectionInitTimeout = DEFAULT_CONNECTION_INIT_TIMEOUT;
+
+    private volatile GraphqlUpgradeSupport upgradeSupport;
+
     /** Constructor. */
     public GraphqlServerAdapter() {
         this(null);
@@ -335,6 +348,37 @@
         this.heartbeatInterval = heartbeatInterval;
     }
 
+    /**
+     * Returns the time allowed to WebSocket clients to send their {@code connection_init} message.
+     *
+     * @return The time allowed to initialize a WebSocket connection.
+     */
+    public Duration getConnectionInitTimeout() {
+        return connectionInitTimeout;
+    }
+
+    /**
+     * Sets the time allowed to WebSocket clients to send their {@code connection_init} message,
+     * after which the connection is closed with code 4408. Defaults to {@link
+     * #DEFAULT_CONNECTION_INIT_TIMEOUT}.
+     *
+     * @param connectionInitTimeout The time allowed to initialize a WebSocket connection.
+     */
+    public void setConnectionInitTimeout(Duration connectionInitTimeout) {
+        checkNotStarted();
+        if (connectionInitTimeout == null
+                || connectionInitTimeout.isZero()
+                || connectionInitTimeout.isNegative()) {
+            throw new IllegalArgumentException("The connection init timeout must be positive");
+        }
+        this.connectionInitTimeout = connectionInitTimeout;
+    }
+
+    /** Returns the counter of active subscriptions, shared by all transports. */
+    AtomicInteger getActiveSubscriptionCounter() {
+        return activeSubscriptions;
+    }
+
     /**
      * Returns the number of active subscriptions.
      *
@@ -357,6 +401,10 @@
             EnumSet<GraphqlTransport> transports = EnumSet.noneOf(GraphqlTransport.class);
             transports.addAll(subscriptionTransports);
             subscriptionTransports = Collections.unmodifiableSet(transports);
+            if (subscriptionTransports.contains(GraphqlTransport.WEBSOCKET)) {
+                // Only loads the optional WebSocket extension when enabled.
+                upgradeSupport = new GraphqlWebSocketSupport(this);
+            }
             super.start();
         }
     }
@@ -371,14 +419,22 @@
         for (GraphqlEventStream stream : List.copyOf(openStreams)) {
             stream.stop();
         }
+        GraphqlUpgradeSupport upgrades = upgradeSupport;
+        if (upgrades != null) {
+            upgrades.stop();
+        }
     }
 
     @Override
     public void handle(Request request, Response response) {
         super.handle(request, response);
         if (!isStarted()) {
             return;
         }
+        GraphqlUpgradeSupport upgrades = upgradeSupport;
+        if (upgrades != null && upgrades.handle(request, response)) {
+            return;
+        }
 
         Method method = request.getMethod();
         boolean get = Method.GET.equals(method);
@@ -433,7 +489,7 @@
         }
 
         // The operation is selected once customized, as the customizer can change it.
-        ExecutionInput executionInput = customize(request, input).build();
+        ExecutionInput executionInput = customize(request, input, null).build();
         OperationDefinition.Operation operation;
         try {
             operation =
@@ -491,17 +547,61 @@
         }
     }
 
-    private ExecutionInput.Builder customize(Request request, ExecutionInput.Builder input) {
+    /**
+     * Customizes and builds the execution input of an operation.
+     *
+     * @param request The request.
+     * @param input The execution input, before customization.
+     * @param initPayload The payload of a WebSocket connection_init message, or null.
+     * @return The execution input.
+     */
+    ExecutionInput prepare(
+            Request request, ExecutionInput.Builder input, Map<String, Object> initPayload) {
+        return customize(request, input, initPayload).build();
+    }
+
+    /**
+     * Indicates if an execution input selects a subscription. Invalid documents are left to the
+     * execution, which reports their errors.
+     *
+     * @param input The execution input.
+     * @return True if the selected operation is a subscription.
+     */
+    boolean isSubscription(ExecutionInput input) {
+        try {
+            return selectOperation(input.getQuery(), input.getOperationName())
+                    == OperationDefinition.Operation.SUBSCRIPTION;
+        } catch (InvalidSyntaxException | RequestException _) {
+            return false;
+        }
+    }
+
+    /**
+     * Executes an operation.
+     *
+     * @param input The execution input, customized.
+     * @return The result.
+     */
+    ExecutionResult execute(ExecutionInput input) {
+        return graphQL.execute(input);
+    }
+
+    private ExecutionInput.Builder customize(
+            Request request, ExecutionInput.Builder input, Map<String, Object> initPayload) {
         BiConsumer<Request, ExecutionInput.Builder> customizer = executionInputCustomizer;
         if (customizer != null) {
             customizer.accept(request, input);
         }
         // graphql-java reads per-execution parser options from the context under their class.
         return input.graphQLContext(
-                builder ->
-                        builder.put(ParserOptions.class, parserOptions)
-                                .put(RestletGraphql.REQUEST, request)
-                                .put(RestletGraphql.CLIENT_INFO, request.getClientInfo()));
+                builder -> {
+                    builder.put(ParserOptions.class, parserOptions)
+                            .put(RestletGraphql.REQUEST, request)
+                            .put(RestletGraphql.CLIENT_INFO, request.getClientInfo());
+                    if (initPayload != null) {
+                        builder.put(RestletGraphql.CONNECTION_INIT_PAYLOAD, initPayload);
+                    }
+                });
     }
 
     /**

org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlTransport.java

@@ -19,5 +19,12 @@ public enum GraphqlTransport {
      * a request accepting {@code text/event-stream} gets one {@code next} event per result and a
      * final {@code complete} event.
      */
-    SSE
+    SSE,
+
+    /**
+     * WebSocket, following the {@code graphql-transport-ws} protocol: the adapter accepts upgrades
+     * offering this subprotocol, on HTTP/1.1 connections only. Requires the {@code
+     * org.restlet.ext.websocket} module.
+     */
+    WEBSOCKET
 }

org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlUpgradeSupport.java

@@ -0,0 +1,32 @@
+/**
+ * Copyright 2005-2026 Qlik
+ *<p>
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *<p>
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.ext.graphql;
+
+import org.restlet.Request;
+import org.restlet.Response;
+
+/**
+ * Handles connection upgrades for {@link GraphqlServerAdapter}. The implementation depends on the
+ * optional WebSocket extension: the adapter only refers to it through this interface, so that the
+ * extension is loaded only when {@link GraphqlTransport#WEBSOCKET} is enabled.
+ */
+interface GraphqlUpgradeSupport {
+
+    /**
+     * Handles a request if it asks for a connection upgrade.
+     *
+     * @param request The request.
+     * @param response The response.
+     * @return True if the request asked for an upgrade and was handled.
+     */
+    boolean handle(Request request, Response response);
+
+    /** Closes the open connections. */
+    void stop();
+}