feat: add GraphQL subscriptions over SSE (G3)

changes.md

@@ -87,7 +87,12 @@ Changes log
           Restlet request and client info through `RestletGraphql` context keys. The client can
           validate documents locally against a `GraphQLSchema` without a network call, bounds
           responses (10 MiB by default, never truncated) and accepts a request customizer for API
-          key headers. Issue #1581.
+          key headers. Subscriptions are served and consumed over Server-Sent Events
+          (`graphql-sse`, distinct connections mode) when `GraphqlTransport.SSE` is enabled: one
+          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.
         - 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/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java

@@ -27,7 +27,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
 import org.restlet.Client;
 import org.restlet.Context;
 import org.restlet.Request;
@@ -95,6 +99,9 @@
 
     private volatile State state = State.NEW;
 
+    private final Set<GraphqlSsePublisher.EventStreamSubscription> subscriptions =
+            ConcurrentHashMap.newKeySet();
+
     /** Constructor. */
     public GraphqlClientAdapter() {
         this(null);
@@ -263,9 +270,16 @@
         state = State.STARTED;
     }
 
-    /** Stops the adapter. The borrowed client is left running. */
+    /**
+     * Stops the adapter. Open subscriptions are cancelled, and their subscribers receive a {@link
+     * GraphqlTransportException}. The borrowed client is left running.
+     */
     public synchronized void stop() {
         state = State.STOPPED;
+        for (GraphqlSsePublisher.EventStreamSubscription subscription :
+                List.copyOf(subscriptions)) {
+            subscription.stop();
+        }
     }
 
     /**
@@ -289,7 +303,7 @@
     public ExecutionResult execute(
             Document document, String operationName, Map<String, Object> variables)
             throws GraphqlTransportException {
         Objects.requireNonNull(document, "document");
         checkStarted();
         ExecutionResult invalid = validate(document);
         return invalid != null
@@ -325,6 +339,128 @@
         return send(document, operationName, variables);
     }
 
+    /**
+     * Subscribes to a parsed operation, usually a subscription, with {@code graphql-sse} in
+     * distinct connections mode.
+     *
+     * @param document The document, printed before sending.
+     * @param operationName The name of the operation to execute, or null.
+     * @param variables The variables, or null.
+     * @return A cold publisher: each subscriber sends its own request on its first demand.
+     * @see #subscribe(String, String, Map)
+     */
+    public Publisher<ExecutionResult> subscribe(
+            Document document, String operationName, Map<String, Object> variables) {
+        Objects.requireNonNull(document, "document");
+        checkStarted();
+        ExecutionResult invalid = validate(document);
+        return invalid != null
+                ? single(invalid)
+                : new GraphqlSsePublisher(
+                        this, AstPrinter.printAst(document), operationName, variables);
+    }
+
+    /**
+     * Subscribes to an operation, usually a subscription, with {@code graphql-sse} in distinct
+     * connections mode. The request is sent with {@code POST}, accepting {@code text/event-stream},
+     * when the first result is requested; each {@code next} event becomes a result, honoring
+     * demand, and the {@code complete} event completes the publisher.
+     *
+     * <p>Request errors are published as a single result with errors. Transport failures, events
+     * larger than {@link #getMaxResponseBytes()} and streams that end without a {@code complete}
+     * event are signalled with a {@link GraphqlTransportException}. Cancelling the subscription
+     * aborts the request.
+     *
+     * @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: each subscriber sends its own request on its first demand.
+     */
+    public Publisher<ExecutionResult> subscribe(
+            String document, String operationName, Map<String, Object> variables) {
+        Objects.requireNonNull(document, "document");
+        checkStarted();
+        if (schema != null) {
+            ExecutionResult invalid;
+            try {
+                invalid = validate(parse(document));
+            } catch (InvalidSyntaxException e) {
+                invalid = failure(List.of(e.toInvalidSyntaxError()));
+            }
+            if (invalid != null) {
+                return single(invalid);
+            }
+        }
+        return new GraphqlSsePublisher(this, document, operationName, variables);
+    }
+
+    /** A publisher of one result, for documents that fail local validation. */
+    private static Publisher<ExecutionResult> single(ExecutionResult result) {
+        return subscriber ->
+                subscriber.onSubscribe(
+                        new Subscription() {
+                            private boolean done;
+
+                            @Override
+                            public void request(long n) {
+                                synchronized (this) {
+                                    if (done) {
+                                        return;
+                                    }
+                                    done = true;
+                                }
+                                if (n <= 0) {
+                                    subscriber.onError(
+                                            new IllegalArgumentException(
+                                                    "The requested number of results must be"
+                                                            + " positive"));
+                                } else {
+                                    subscriber.onNext(result);
+                                    subscriber.onComplete();
+                                }
+                            }
+
+                            @Override
+                            public synchronized void cancel() {
+                                done = true;
+                            }
+                        });
+    }
+
+    /** Registers an open subscription. Returns false if the adapter is not started. */
+    synchronized boolean register(GraphqlSsePublisher.EventStreamSubscription subscription) {
+        if (state != State.STARTED) {
+            return false;
+        }
+        subscriptions.add(subscription);
+        return true;
+    }
+
+    void unregister(GraphqlSsePublisher.EventStreamSubscription subscription) {
+        subscriptions.remove(subscription);
+    }
+
+    /**
+     * Fills the HTTP request of an operation sent for an event stream.
+     *
+     * @param request The new request to fill.
+     * @param document The document.
+     * @param operationName The name of the operation to execute, or null.
+     * @param variables The variables, or null.
+     * @return The request, customized.
+     */
+    Request createEventStreamRequest(
+            Request request, String document, String operationName, Map<String, Object> variables) {
+        createRequest(request, document, operationName, variables);
+        request.getClientInfo()
+                .getAcceptedMediaTypes()
+                .add(new Preference<>(MediaType.TEXT_EVENT_STREAM, 1F));
+        request.getClientInfo()
+                .getAcceptedMediaTypes()
+                .add(new Preference<>(MediaType.APPLICATION_JSON, 0.5F));
+        return request;
+    }
+
     private Document parse(String document) {
         return Parser.parse(
                 ParserEnvironment.newParserEnvironment()
@@ -356,6 +492,11 @@
      * @return The request, customized.
      */
     Request createRequest(String document, String operationName, Map<String, Object> variables) {
+        return createRequest(new Request(), document, operationName, variables);
+    }
+
+    private Request createRequest(
+            Request request, String document, String operationName, Map<String, Object> variables) {
         Map<String, Object> body = new LinkedHashMap<>();
         body.put("query", document);
         if (operationName != null) {
@@ -365,7 +506,8 @@
             body.put("variables", variables);
         }
 
-        Request request = new Request(Method.POST, endpoint);
+        request.setMethod(Method.POST);
+        request.setResourceRef(endpoint);
         request.setEntity(
                 new StringRepresentation(
                         GraphqlJson.writeString(body),
@@ -401,7 +543,14 @@
         }
     }
 
-    private ExecutionResult toResult(Response response) throws GraphqlTransportException {
+    /**
+     * Converts a response with a single GraphQL result.
+     *
+     * @param response The response.
+     * @return The result.
+     * @throws GraphqlTransportException If the response is not a GraphQL result.
+     */
+    ExecutionResult toResult(Response response) throws GraphqlTransportException {
         Status status = response.getStatus();
         if (status.isConnectorError()) {
             throw new GraphqlTransportException(