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

changes.md

@@ -70,11 +70,14 @@ Changes log
           `WebSocketUpgrader` (or `WebSocketUpgrade.accept`) accepts upgrades that went through
           the normal routing, filters and authenticators, and runs a `WebSocketHandler` on its own
           virtual thread with a blocking `WebSocketSession` (`receive`, `sendText`, `sendBinary`,
-          `close`). `WebSocketClient` opens client sessions on "ws" and "wss" URIs. The server
+          `close`). `WebSocketClient` opens client sessions on "ws" and "wss" URIs; it is configured
+          with setters, which are rejected while it is started, and can be stopped, reconfigured
+          and started again (`connect` starts a stopped client). The server
           answers pings, pings idle sessions, and supports the "websocket.maxSessions",
           "websocket.maxMessageSize", "websocket.maxFrameSize" and "websocket.compression"
           parameters. Server and client support "permessage-deflate" compression (RFC 7692),
-          enabled by default.
+          enabled by default. `WebSocketClient.setClient` shares the I/O threads of a borrowed,
+          started Netty HTTP `Client` instead of creating its own.
         - Added the experimental `org.restlet.ext.graphql` extension, on graphql-java.
           `GraphqlServerAdapter` serves a `graphql.GraphQL` instance as a Restlet, following the
           GraphQL over HTTP draft (`application/graphql-response+json` and `application/json`);
@@ -96,10 +99,14 @@ Changes log
           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.
+          loaded only then. `GraphqlClientAdapter.setSubscriptionTransport(WEBSOCKET)` multiplexes
+          subscriptions over one `graphql-transport-ws` connection sharing the borrowed client's
+          event loop. 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.
+          `NettyClientHelper.getEventLoopGroup()` lets extensions share the event loop of the Netty
+          HTTP client connector.
         - Added `Response.getProtocol()` and `setProtocol()`: the protocol, with its version,
           used by the server to reply (for example HTTP/1.1, HTTP/2.0 or HTTP/3.0), set by the HTTP
           client connector. When unknown, the protocol of the request is returned.
@@ -119,6 +126,9 @@ Changes log
         - Fixed the OpenAPI extension build path for Commons Lang types inherited from Swagger.
         - Slf4j bridge reports wrong line number and caller class in log4j. Issue #233.
         - When GET is an allowed method of a server resource, HEAD is now also listed (e.g. in the Allow header of a 405 response). Issue #266.
+        - `Client.setContext()` now stores the connector helper in the "org.restlet.engine.helper"
+          context attribute, as the constructors do, so clients added with
+          `Component.getClients().add(Protocol)` expose it too.
     - Misc
         - Fixed JDK HTTP client accessibility in core and WebSocket tests in Eclipse and VS Code,
           without adding a production module dependency.

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

@@ -99,6 +99,12 @@ private enum State {
 
     private volatile State state = State.NEW;
 
+    private volatile GraphqlTransport subscriptionTransport = GraphqlTransport.SSE;
+
+    private volatile Map<String, Object> connectionInitPayload;
+
+    private volatile GraphqlSubscriptionTransport webSocketTransport;
+
     private final Set<GraphqlSsePublisher.EventStreamSubscription> subscriptions =
             ConcurrentHashMap.newKeySet();
 
@@ -250,6 +256,50 @@ public void setRequestCustomizer(Consumer<Request> requestCustomizer) {
         this.requestCustomizer = requestCustomizer;
     }
 
+    /**
+     * Returns the transport of subscriptions.
+     *
+     * @return The transport of subscriptions.
+     */
+    public GraphqlTransport getSubscriptionTransport() {
+        return subscriptionTransport;
+    }
+
+    /**
+     * Sets the transport of subscriptions. Defaults to {@link GraphqlTransport#SSE}, with one
+     * request per subscription. With {@link GraphqlTransport#WEBSOCKET}, subscriptions are
+     * multiplexed over one {@code graphql-transport-ws} connection, opened on the first demand at
+     * the endpoint URI with the "ws" or "wss" scheme. It shares the event loop of the borrowed
+     * client, which must then be a started Netty HTTP client, and requires the {@code
+     * org.restlet.ext.websocket} module. The request customizer adds handshake headers.
+     *
+     * @param subscriptionTransport The transport of subscriptions.
+     */
+    public void setSubscriptionTransport(GraphqlTransport subscriptionTransport) {
+        checkNew();
+        this.subscriptionTransport = Objects.requireNonNull(subscriptionTransport);
+    }
+
+    /**
+     * Returns the payload of the {@code connection_init} message of WebSocket connections.
+     *
+     * @return The payload, or null.
+     */
+    public Map<String, Object> getConnectionInitPayload() {
+        return connectionInitPayload;
+    }
+
+    /**
+     * Sets the payload of the {@code connection_init} message of WebSocket connections, for example
+     * an authentication token.
+     *
+     * @param connectionInitPayload The payload, or null.
+     */
+    public void setConnectionInitPayload(Map<String, Object> connectionInitPayload) {
+        checkNew();
+        this.connectionInitPayload = connectionInitPayload;
+    }
+
     /**
      * Validates and freezes the configuration.
      *
@@ -267,6 +317,14 @@ public synchronized void start() {
         if (parserOptions == null) {
             parserOptions = ParserOptions.getDefaultOperationParserOptions();
         }
+        if (subscriptionTransport == GraphqlTransport.WEBSOCKET) {
+            try {
+                // Only loads the optional WebSocket extension when selected.
+                webSocketTransport = new GraphqlWebSocketTransport(this);
+            } catch (IOException e) {
+                throw new IllegalStateException("Unable to start the WebSocket client", e);
+            }
+        }
         state = State.STARTED;
     }
 
@@ -280,6 +338,10 @@ public synchronized void stop() {
                 List.copyOf(subscriptions)) {
             subscription.stop();
         }
+        GraphqlSubscriptionTransport transport = webSocketTransport;
+        if (transport != null) {
+            transport.stop();
+        }
     }
 
     /**
@@ -356,8 +418,7 @@ public Publisher<ExecutionResult> subscribe(
         ExecutionResult invalid = validate(document);
         return invalid != null
                 ? single(invalid)
-                : new GraphqlSsePublisher(
-                        this, AstPrinter.printAst(document), operationName, variables);
+                : publisher(AstPrinter.printAst(document), operationName, variables);
     }
 
     /**
@@ -391,7 +452,15 @@ public Publisher<ExecutionResult> subscribe(
                 return single(invalid);
             }
         }
-        return new GraphqlSsePublisher(this, document, operationName, variables);
+        return publisher(document, operationName, variables);
+    }
+
+    private Publisher<ExecutionResult> publisher(
+            String document, String operationName, Map<String, Object> variables) {
+        GraphqlSubscriptionTransport transport = webSocketTransport;
+        return transport != null
+                ? transport.subscribe(document, operationName, variables)
+                : new GraphqlSsePublisher(this, document, operationName, variables);
     }
 
     /** A publisher of one result, for documents that fail local validation. */

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

@@ -0,0 +1,36 @@
+/**
+ * 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 graphql.ExecutionResult;
+import java.util.Map;
+import org.reactivestreams.Publisher;
+
+/**
+ * A subscription transport of {@link GraphqlClientAdapter} other than SSE. 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
+ * selected.
+ */
+interface GraphqlSubscriptionTransport {
+
+    /**
+     * Subscribes to an operation.
+     *
+     * @param document The document.
+     * @param operationName The name of the operation to execute, or null.
+     * @param variables The variables, or null.
+     * @return A cold publisher of the results.
+     */
+    Publisher<ExecutionResult> subscribe(
+            String document, String operationName, Map<String, Object> variables);
+
+    /** Fails the open subscriptions and closes the connection. */
+    void stop();
+}