From 06e49e786953b3b46fa5d8a2bdeec19a35f4863f Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 22:50:23 -0400 Subject: [PATCH 1/2] feat: add GraphQL subscriptions over SSE (G3) graphql-sse in distinct connections mode, on core SSE: - Server: GraphqlTransport.SSE enables event streams for requests accepting text/event-stream: subscriptions (also over GET when it is enabled), and queries and mutations as one next event. Results are requested one at a time and written as next events, then complete. Request errors are 400 with a JSON body. An initial comment sends the response head at once; heartbeats (15 s by default) keep idle streams open and detect disconnected clients. A failed write cancels the subscription. - Limits: 64 active subscriptions per adapter (503 beyond), 1 MiB of queued bytes per subscription (the stream ends with an error and the subscription is cancelled). Stopping the adapter ends its open streams so that the server connector can shut down gracefully. - Client: subscribe() returns a cold Publisher; each subscriber sends its own request on its first demand, from a virtual thread that emits all the signals. Demand is honored, cancel() aborts the request, events are bounded by maxResponseBytes, request errors are published as a single result, and a stream ending without a complete event is a GraphqlTransportException. stop() fails the open subscriptions. Refs #1585 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 7 +- .../ext/graphql/GraphqlClientAdapter.java | 146 +++++- .../ext/graphql/GraphqlServerAdapter.java | 213 +++++++- .../ext/graphql/GraphqlSsePublisher.java | 321 ++++++++++++ .../restlet/ext/graphql/GraphqlTransport.java | 23 + .../graphql/internal/GraphqlEventStream.java | 308 ++++++++++++ .../ext/graphql/GraphqlSseTestCase.java | 476 ++++++++++++++++++ .../ext/graphql/GraphqlTestSchemas.java | 8 +- .../restlet/ext/graphql/TickPublisher.java | 13 +- 9 files changed, 1501 insertions(+), 14 deletions(-) create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlTransport.java create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java create mode 100644 org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseTestCase.java diff --git a/changes.md b/changes.md index 2ff021aef3..b86018b38d 100644 --- a/changes.md +++ b/changes.md @@ -147,7 +147,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` + 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. diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java index 6457e1dc22..7b1116b1df 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java +++ b/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 enum State { private volatile State state = State.NEW; + private final Set subscriptions = + ConcurrentHashMap.newKeySet(); + /** Constructor. */ public GraphqlClientAdapter() { this(null); @@ -263,9 +270,16 @@ public synchronized void start() { 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(); + } } /** @@ -325,6 +339,127 @@ public ExecutionResult execute( 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 subscribe( + Document document, String operationName, Map 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. + * + *

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 subscribe( + String document, String operationName, Map 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 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); + } + + /** + * Creates the HTTP request of an operation sent for an event stream. + * + * @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( + String document, String operationName, Map variables) { + Request request = createRequest(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() @@ -401,7 +536,14 @@ private ExecutionResult send( } } - 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( diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java index 5c8afca210..04b3894af9 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java @@ -21,10 +21,17 @@ import graphql.parser.ParserOptions; import java.io.IOException; import java.io.InputStream; +import java.time.Duration; +import java.util.Collections; +import java.util.EnumSet; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; +import org.reactivestreams.Publisher; import org.restlet.Context; import org.restlet.Request; import org.restlet.Response; @@ -33,10 +40,13 @@ import org.restlet.data.Form; import org.restlet.data.MediaType; import org.restlet.data.Method; +import org.restlet.data.Preference; import org.restlet.data.Status; +import org.restlet.ext.graphql.internal.GraphqlEventStream; import org.restlet.ext.graphql.internal.GraphqlJson; import org.restlet.ext.graphql.internal.GraphqlMediaTypes; import org.restlet.representation.ByteArrayRepresentation; +import org.restlet.representation.EventStreamRepresentation; import org.restlet.representation.Representation; /** @@ -65,6 +75,14 @@ * 200}, even with field errors. * * + *

Subscriptions are rejected unless a transport is enabled with {@link + * #getSubscriptionTransports()}. With {@link GraphqlTransport#SSE}, a request accepting {@code + * text/event-stream} gets a {@code graphql-sse} event stream in distinct connections mode, for + * subscriptions as well as queries and mutations, including subscriptions over {@code GET} when it + * is enabled. Results are requested one at a time and written as {@code next} events, followed by a + * {@code complete} event. A client disconnect, detected on write or by heartbeats, cancels the + * subscription, and stopping the adapter ends its open streams. + * *

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 @@ -80,6 +98,15 @@ public class GraphqlServerAdapter extends Restlet { /** The default maximum size of a request body: 1 MiB. */ public static final long DEFAULT_MAX_REQUEST_BYTES = 1L << 20; + /** The default maximum number of active subscriptions of an adapter: 64. */ + public static final int DEFAULT_MAX_SUBSCRIPTIONS = 64; + + /** The default maximum number of bytes queued for a subscription: 1 MiB. */ + public static final long DEFAULT_MAX_QUEUED_BYTES = 1L << 20; + + /** The default interval of event stream heartbeats: 15 seconds. */ + public static final Duration DEFAULT_HEARTBEAT_INTERVAL = Duration.ofSeconds(15); + private volatile GraphQL graphQL; private volatile long maxRequestBytes = DEFAULT_MAX_REQUEST_BYTES; @@ -90,6 +117,19 @@ public class GraphqlServerAdapter extends Restlet { private volatile BiConsumer executionInputCustomizer; + private volatile Set subscriptionTransports = + EnumSet.noneOf(GraphqlTransport.class); + + private volatile int maxSubscriptions = DEFAULT_MAX_SUBSCRIPTIONS; + + private volatile long maxQueuedBytes = DEFAULT_MAX_QUEUED_BYTES; + + private volatile Duration heartbeatInterval = DEFAULT_HEARTBEAT_INTERVAL; + + private final AtomicInteger activeSubscriptions = new AtomicInteger(); + + private final Set openStreams = ConcurrentHashMap.newKeySet(); + /** Constructor. */ public GraphqlServerAdapter() { this(null); @@ -209,6 +249,101 @@ public void setExecutionInputCustomizer( this.executionInputCustomizer = executionInputCustomizer; } + /** + * Returns the transports enabled for subscriptions. Empty by default: subscriptions are then + * rejected. The set can be modified until the adapter starts. + * + * @return The transports enabled for subscriptions. + */ + public Set getSubscriptionTransports() { + return subscriptionTransports; + } + + /** + * Returns the maximum number of active subscriptions of this adapter. + * + * @return The maximum number of active subscriptions. + */ + public int getMaxSubscriptions() { + return maxSubscriptions; + } + + /** + * Sets the maximum number of active subscriptions of this adapter. Further subscriptions get + * {@code 503}. Defaults to {@link #DEFAULT_MAX_SUBSCRIPTIONS}. The {@code + * server.maxEventStreams} parameter of the server connector still applies. + * + * @param maxSubscriptions The maximum number of active subscriptions. + */ + public void setMaxSubscriptions(int maxSubscriptions) { + checkNotStarted(); + if (maxSubscriptions <= 0) { + throw new IllegalArgumentException( + "The maximum number of subscriptions must be positive"); + } + this.maxSubscriptions = maxSubscriptions; + } + + /** + * Returns the maximum number of bytes of results received from a subscription but not yet + * written. + * + * @return The maximum number of queued bytes per subscription. + */ + public long getMaxQueuedBytes() { + return maxQueuedBytes; + } + + /** + * Sets the maximum number of bytes of results received from a subscription but not yet written. + * Results are requested one at a time, so this bounds the size of a result and protects against + * publishers that ignore demand. Beyond it, the subscription is cancelled and the stream ends + * with an error. Defaults to {@link #DEFAULT_MAX_QUEUED_BYTES}. + * + * @param maxQueuedBytes The maximum number of queued bytes per subscription. + */ + public void setMaxQueuedBytes(long maxQueuedBytes) { + checkNotStarted(); + if (maxQueuedBytes <= 0) { + throw new IllegalArgumentException("The maximum queued bytes must be positive"); + } + this.maxQueuedBytes = maxQueuedBytes; + } + + /** + * Returns the interval of event stream heartbeats. + * + * @return The interval of event stream heartbeats, or null if disabled. + */ + public Duration getHeartbeatInterval() { + return heartbeatInterval; + } + + /** + * Sets the interval of comment heartbeats on event streams. They keep idle subscriptions open + * through proxies and detect disconnected clients. Defaults to {@link + * #DEFAULT_HEARTBEAT_INTERVAL}. + * + * @param heartbeatInterval The interval of event stream heartbeats, or null to disable them. + */ + public void setHeartbeatInterval(Duration heartbeatInterval) { + checkNotStarted(); + if (heartbeatInterval != null + && (heartbeatInterval.isZero() || heartbeatInterval.isNegative())) { + throw new IllegalArgumentException("The heartbeat interval must be positive"); + } + this.heartbeatInterval = heartbeatInterval; + } + + /** + * Returns the number of active subscriptions. + * + * @return The number of active subscriptions. + */ + public int getActiveSubscriptions() { + return activeSubscriptions.get(); + } + @Override public synchronized void start() throws Exception { if (isStopped()) { @@ -218,10 +353,24 @@ public synchronized void start() throws Exception { if (parserOptions == null) { parserOptions = ParserOptions.getDefaultOperationParserOptions(); } + subscriptionTransports = + Collections.unmodifiableSet(EnumSet.copyOf(subscriptionTransports)); super.start(); } } + /** + * Stops the adapter. Open subscription streams are cancelled and end with an error and a {@code + * complete} event, so that the server connector can shut down gracefully. + */ + @Override + public synchronized void stop() throws Exception { + super.stop(); + for (GraphqlEventStream stream : List.copyOf(openStreams)) { + stream.stop(); + } + } + @Override public void handle(Request request, Response response) { super.handle(request, response); @@ -238,11 +387,18 @@ public void handle(Request request, Response response) { // The response media type is negotiated, which caches must know of. response.getDimensions().add(Dimension.MEDIA_TYPE); - MediaType responseType = - GraphqlMediaTypes.negotiate(request.getClientInfo().getAcceptedMediaTypes()); + List> accepted = request.getClientInfo().getAcceptedMediaTypes(); + boolean sse = + subscriptionTransports.contains(GraphqlTransport.SSE) + && GraphqlMediaTypes.quality(accepted, MediaType.TEXT_EVENT_STREAM, true) + > 0F; + MediaType responseType = GraphqlMediaTypes.negotiate(accepted); if (responseType == null) { - response.setStatus(Status.CLIENT_ERROR_NOT_ACCEPTABLE); - return; + if (!sse) { + response.setStatus(Status.CLIENT_ERROR_NOT_ACCEPTABLE); + return; + } + responseType = MediaType.APPLICATION_JSON; } Map params; @@ -288,19 +444,49 @@ public void handle(Request request, Response response) { return; } - if (get && operation != OperationDefinition.Operation.QUERY) { + boolean subscription = operation == OperationDefinition.Operation.SUBSCRIPTION; + if (get && operation != OperationDefinition.Operation.QUERY && !(sse && subscription)) { methodNotAllowed(response, false); return; } - if (operation == OperationDefinition.Operation.SUBSCRIPTION) { + if (subscription && !sse) { write( response, responseType, - failure(error("Subscriptions are not supported by this endpoint"))); + failure(error("Subscriptions require an event stream on this endpoint"))); + return; + } + if (subscription && activeSubscriptions.get() >= maxSubscriptions) { + writeEntity(response, responseType, failure(error("Too many active subscriptions"))); + response.setStatus(Status.SERVER_ERROR_SERVICE_UNAVAILABLE); return; } - write(response, responseType, graphQL.execute(executionInput)); + ExecutionResult result = graphQL.execute(executionInput); + if (!sse || !result.isDataPresent()) { + // Without an event stream, or when execution didn't start: a single JSON result. + write(response, responseType, result, sse); + } else if (result.getData() instanceof Publisher publisher) { + @SuppressWarnings("unchecked") + Publisher results = (Publisher) publisher; + response.setEntity( + new EventStreamRepresentation( + new GraphqlEventStream( + results, + activeSubscriptions, + openStreams, + this::isStopped, + maxSubscriptions, + maxQueuedBytes, + getLogger()), + heartbeatInterval)); + response.setStatus(Status.SUCCESS_OK); + } else { + response.setEntity( + new EventStreamRepresentation( + new GraphqlEventStream(result), heartbeatInterval)); + response.setStatus(Status.SUCCESS_OK); + } } private ExecutionInput.Builder customize(Request request, ExecutionInput.Builder input) { @@ -464,10 +650,19 @@ private static void requestError(Response response, MediaType responseType, Stri * reported with {@code 400} in {@code application/graphql-response+json} only. */ private static void write(Response response, MediaType responseType, ExecutionResult result) { + write(response, responseType, result, false); + } + + /** + * Writes a result. Event stream requests that fail before execution always get {@code 400}, as + * in {@code graphql-sse}. + */ + private static void write( + Response response, MediaType responseType, ExecutionResult result, boolean sse) { writeEntity(response, responseType, result); boolean requestError = !result.isDataPresent() - && GraphqlMediaTypes.GRAPHQL_RESPONSE_JSON.equals(responseType); + && (sse || GraphqlMediaTypes.GRAPHQL_RESPONSE_JSON.equals(responseType)); response.setStatus(requestError ? Status.CLIENT_ERROR_BAD_REQUEST : Status.SUCCESS_OK); } diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java new file mode 100644 index 0000000000..319344d85f --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java @@ -0,0 +1,321 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql; + +import graphql.ExecutionResult; +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.Map; +import java.util.Objects; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.MediaType; +import org.restlet.data.ServerSentEvent; +import org.restlet.ext.graphql.internal.GraphqlJson; +import org.restlet.representation.EventStreamReader; +import org.restlet.representation.Representation; + +/** + * A cold publisher of the results of an operation sent with {@code graphql-sse} in distinct + * connections mode. Each subscriber opens its own request, on the first demand. One virtual thread + * per subscription reads the events and emits all the signals. + */ +final class GraphqlSsePublisher implements Publisher { + + private final GraphqlClientAdapter adapter; + + private final String document; + + private final String operationName; + + private final Map variables; + + GraphqlSsePublisher( + GraphqlClientAdapter adapter, + String document, + String operationName, + Map variables) { + this.adapter = adapter; + this.document = document; + this.operationName = operationName; + this.variables = variables; + } + + @Override + public void subscribe(Subscriber subscriber) { + Objects.requireNonNull(subscriber, "subscriber"); + EventStreamSubscription subscription = new EventStreamSubscription(subscriber); + subscriber.onSubscribe(subscription); + if (!adapter.register(subscription)) { + subscription.stop(); + } + } + + /** The subscription of one subscriber. */ + final class EventStreamSubscription implements Subscription { + + private final Subscriber subscriber; + + private long demand; + + private boolean started; + + private boolean invalidDemand; + + private volatile boolean cancelled; + + private volatile boolean stopped; + + private volatile Request request; + + EventStreamSubscription(Subscriber subscriber) { + this.subscriber = subscriber; + } + + @Override + public void request(long n) { + synchronized (this) { + if (cancelled) { + return; + } + if (n <= 0) { + invalidDemand = true; + } else { + demand = demand + n < 0 ? Long.MAX_VALUE : demand + n; + } + notifyAll(); + if (started) { + return; + } + started = true; + } + Thread.ofVirtual().name("restlet-graphql-sse").start(this::run); + } + + @Override + public void cancel() { + synchronized (this) { + if (cancelled) { + return; + } + cancelled = true; + notifyAll(); + } + Request current = request; + if (current != null) { + current.abort(); + } + adapter.unregister(this); + } + + /** Stops the subscription because the adapter stops, signalling an error. */ + void stop() { + boolean signal; + synchronized (this) { + if (cancelled) { + return; + } + stopped = true; + // Without a reader thread, signal from here and prevent one from starting. + signal = !started; + started = true; + } + if (signal) { + subscriber.onError(stoppedError()); + } + cancel(); + } + + private GraphqlTransportException stoppedError() { + return new GraphqlTransportException( + "The GraphQL client was stopped", null, null, null); + } + + private void run() { + Response response = null; + try { + Request current = + adapter.createEventStreamRequest(document, operationName, variables); + request = current; + if (cancelled) { + terminateIfStopped(); + return; + } + response = adapter.getClient().handle(current); + if (cancelled) { + terminateIfStopped(); + return; + } + Representation entity = response.getEntity(); + if (entity == null + || entity.getMediaType() == null + || !MediaType.TEXT_EVENT_STREAM.equals(entity.getMediaType(), true)) { + ExecutionResult result = adapter.toResult(response); + if (awaitDemand()) { + subscriber.onNext(result); + subscriber.onComplete(); + } + return; + } + readEvents(entity.getStream()); + } catch (GraphqlTransportException e) { + fail(e); + } catch (IOException e) { + // Only thrown while reading the event stream of a response. + fail( + new GraphqlTransportException( + "Unable to read the GraphQL event stream", + response.getStatus(), + null, + e)); + } catch (RuntimeException e) { + fail(e); + } finally { + if (response != null && response.getEntity() != null) { + response.getEntity().release(); + } + adapter.unregister(this); + } + } + + private void readEvents(InputStream stream) throws IOException { + BoundedInputStream input = + new BoundedInputStream(stream, adapter.getMaxResponseBytes()); + try (EventStreamReader reader = new EventStreamReader(input)) { + while (!cancelled) { + ServerSentEvent event = reader.readEvent(); + input.resetCount(); + if (event == null) { + if (!cancelled) { + throw new GraphqlTransportException( + "The GraphQL event stream ended without a complete event", + null, + null, + null); + } + } else if ("next".equals(event.getEvent())) { + Map result = + GraphqlJson.asObject(GraphqlJson.parse(event.getData())); + if (result == null) { + throw new GraphqlTransportException( + "Invalid next event", null, event.getData(), null); + } + if (!awaitDemand()) { + break; + } + subscriber.onNext(ExecutionResult.fromSpecification(result)); + } else if ("complete".equals(event.getEvent())) { + if (!cancelled) { + subscriber.onComplete(); + } + return; + } + } + terminateIfStopped(); + } catch (GraphqlJson.InvalidJsonException e) { + throw new GraphqlTransportException("Invalid next event", null, null, e); + } + } + + /** Waits for demand. Returns false if cancelled meanwhile. */ + private boolean awaitDemand() { + synchronized (this) { + while (!cancelled && demand == 0 && !invalidDemand) { + try { + wait(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + if (cancelled) { + return false; + } + if (invalidDemand) { + cancelled = true; + subscriber.onError( + new IllegalArgumentException( + "The requested number of results must be positive")); + return false; + } + if (demand != Long.MAX_VALUE) { + demand--; + } + return true; + } + } + + private void terminateIfStopped() { + if (stopped) { + subscriber.onError(stoppedError()); + } + } + + private void fail(Throwable error) { + if (stopped) { + subscriber.onError(stoppedError()); + } else if (!cancelled) { + cancelled = true; + subscriber.onError(error); + } + } + } + + /** + * Limits the bytes read for each event. The event reader buffers ahead, so the limit is checked + * with a margin of one buffer. + */ + private static final class BoundedInputStream extends FilterInputStream { + + private static final int READ_AHEAD = 8192; + + private final long limit; + + private long count; + + BoundedInputStream(InputStream in, long limit) { + super(in); + this.limit = limit; + } + + void resetCount() { + count = 0; + } + + @Override + public int read() throws IOException { + int b = super.read(); + if (b >= 0) { + add(1); + } + return b; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int n = super.read(b, off, len); + if (n > 0) { + add(n); + } + return n; + } + + private void add(int n) throws GraphqlTransportException { + count += n; + if (count > limit + READ_AHEAD) { + throw new GraphqlTransportException( + "A GraphQL event exceeds " + limit + " bytes", null, null, null); + } + } + } +} diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlTransport.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlTransport.java new file mode 100644 index 0000000000..3f4111c3da --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlTransport.java @@ -0,0 +1,23 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql; + +/** + * Transports of GraphQL subscriptions, enabled with {@link + * GraphqlServerAdapter#getSubscriptionTransports()}. + */ +public enum GraphqlTransport { + + /** + * Server-Sent Events, following the {@code graphql-sse} protocol in distinct connections mode: + * a request accepting {@code text/event-stream} gets one {@code next} event per result and a + * final {@code complete} event. + */ + SSE +} diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java new file mode 100644 index 0000000000..4c52d08035 --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java @@ -0,0 +1,308 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql.internal; + +import graphql.ExecutionResult; +import graphql.GraphqlErrorBuilder; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Set; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.data.ServerSentEvent; +import org.restlet.representation.EventStreamRepresentation.EventSink; +import org.restlet.representation.EventStreamRepresentation.EventWriter; + +/** + * Writes the results of an operation as a {@code graphql-sse} event stream in distinct connections + * mode: one {@code next} event per result, then a {@code complete} event. + * + *

A subscription is consumed one result at a time: the next one is requested once the previous + * event is written, so the connector's flow control applies to the publisher. Results that arrive + * without being requested are queued up to a byte limit, beyond which the subscription is + * cancelled. A failed write, such as a client disconnect, cancels the subscription. + */ +public final class GraphqlEventStream implements EventWriter { + + private static final ServerSentEvent COMPLETE = + ServerSentEvent.builder().event("complete").data("").build(); + + private static final ServerSentEvent OPEN = ServerSentEvent.builder().comment("").build(); + + private final ExecutionResult result; + + private final Publisher publisher; + + private final AtomicInteger activeSubscriptions; + + private final Set openStreams; + + private final BooleanSupplier endpointStopped; + + private volatile QueueingSubscriber subscriber; + + private volatile boolean stopped; + + private final int maxSubscriptions; + + private final long maxQueuedBytes; + + private final Logger logger; + + /** + * Creates the stream of a single result. + * + * @param result The result of a query or mutation. + */ + public GraphqlEventStream(ExecutionResult result) { + this.result = result; + this.publisher = null; + this.activeSubscriptions = null; + this.openStreams = null; + this.endpointStopped = null; + this.maxSubscriptions = 0; + this.maxQueuedBytes = 0; + this.logger = null; + } + + /** + * Creates the stream of a subscription. + * + * @param publisher The publisher of the subscription results. + * @param activeSubscriptions The counter of active subscriptions, shared by an adapter. + * @param openStreams The open streams of an adapter, stopped with {@link #stop()}. + * @param endpointStopped Indicates if the adapter is stopped, checked once the stream is open. + * @param maxSubscriptions The maximum number of active subscriptions. + * @param maxQueuedBytes The maximum number of bytes of results received but not yet written. + * @param logger The logger of publisher failures. + */ + public GraphqlEventStream( + Publisher publisher, + AtomicInteger activeSubscriptions, + Set openStreams, + BooleanSupplier endpointStopped, + int maxSubscriptions, + long maxQueuedBytes, + Logger logger) { + this.result = null; + this.publisher = publisher; + this.activeSubscriptions = activeSubscriptions; + this.openStreams = openStreams; + this.endpointStopped = endpointStopped; + this.maxSubscriptions = maxSubscriptions; + this.maxQueuedBytes = maxQueuedBytes; + this.logger = logger; + } + + /** + * Returns the {@code next} event of a result. + * + * @param result The result. + * @return The {@code next} event. + */ + public static ServerSentEvent next(ExecutionResult result) { + return next(GraphqlJson.writeString(result.toSpecification())); + } + + private static ServerSentEvent next(String json) { + return ServerSentEvent.builder().event("next").data(json).build(); + } + + private static ServerSentEvent error(String message) { + return next( + ExecutionResult.newExecutionResult() + .errors( + List.of( + GraphqlErrorBuilder.newError() + .message("%s", message) + .build())) + .build()); + } + + /** + * Stops the stream, for example because its adapter stops: the subscription is cancelled and + * the stream ends with an error and a {@code complete} event. + */ + public void stop() { + stopped = true; + QueueingSubscriber current = subscriber; + if (current != null) { + current.fail(new Failure("The GraphQL endpoint is stopping", null)); + } + } + + @Override + public void write(EventSink sink) throws IOException { + if (publisher == null) { + sink.send(next(result)); + sink.send(COMPLETE); + return; + } + + if (activeSubscriptions.incrementAndGet() > maxSubscriptions) { + activeSubscriptions.decrementAndGet(); + sink.send(error("Too many active subscriptions")); + sink.send(COMPLETE); + return; + } + // An empty comment sends the response head at once, before the first result. + sink.send(OPEN); + QueueingSubscriber subscriber = new QueueingSubscriber(maxQueuedBytes); + this.subscriber = subscriber; + openStreams.add(this); + try { + // Registered before checking, so that a concurrent stop of the adapter sees it. + if (stopped || endpointStopped.getAsBoolean()) { + subscriber.fail(new Failure("The GraphQL endpoint is stopping", null)); + } else { + publisher.subscribe(subscriber); + } + while (true) { + Object signal = subscriber.take(); + if (signal instanceof Next next) { + sink.send(next(next.json())); + subscriber.request(); + } else if (signal instanceof Failure failure) { + if (failure.cause() != null) { + logger.log(Level.WARNING, "GraphQL subscription failed", failure.cause()); + } + sink.send(error(failure.message())); + sink.send(COMPLETE); + return; + } else { + sink.send(COMPLETE); + return; + } + } + } catch (InterruptedException e) { + // Interrupted by a failed heartbeat: the client is gone. + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Event stream interrupted"); + } finally { + subscriber.cancel(); + openStreams.remove(this); + activeSubscriptions.decrementAndGet(); + } + } + + /** A serialized result and its size in bytes. */ + private record Next(String json, long size) {} + + /** A terminal failure of the stream. */ + private record Failure(String message, Throwable cause) {} + + /** The completion of the stream. */ + private static final Object COMPLETED = new Object(); + + /** Queues the serialized results of a subscription for the writer thread. */ + private static final class QueueingSubscriber implements Subscriber { + + private final LinkedBlockingQueue signals = new LinkedBlockingQueue<>(); + + private final long maxQueuedBytes; + + private long queuedBytes; + + private volatile Subscription subscription; + + private volatile boolean cancelled; + + private boolean terminated; + + QueueingSubscriber(long maxQueuedBytes) { + this.maxQueuedBytes = maxQueuedBytes; + } + + @Override + public void onSubscribe(Subscription subscription) { + this.subscription = subscription; + if (cancelled) { + subscription.cancel(); + } else { + subscription.request(1); + } + } + + @Override + public void onNext(ExecutionResult item) { + String json = GraphqlJson.writeString(item.toSpecification()); + long size = json.getBytes(StandardCharsets.UTF_8).length; + synchronized (this) { + if (terminated) { + return; + } + queuedBytes += size; + if (queuedBytes > maxQueuedBytes) { + terminated = true; + signals.add( + new Failure("The subscription exceeded its queued bytes limit", null)); + cancel(); + return; + } + signals.add(new Next(json, size)); + } + } + + @Override + public void onError(Throwable throwable) { + terminate(new Failure("The subscription failed", throwable)); + } + + @Override + public void onComplete() { + terminate(COMPLETED); + } + + /** Ends the stream with a failure and cancels the subscription. */ + void fail(Failure failure) { + terminate(failure); + cancel(); + } + + private synchronized void terminate(Object signal) { + if (!terminated) { + terminated = true; + signals.add(signal); + } + } + + Object take() throws InterruptedException { + Object signal = signals.take(); + if (signal instanceof Next next) { + synchronized (this) { + queuedBytes -= next.size(); + } + } + return signal; + } + + void request() { + Subscription current = subscription; + if (current != null && !cancelled) { + current.request(1); + } + } + + void cancel() { + cancelled = true; + Subscription current = subscription; + if (current != null) { + current.cancel(); + } + } + } +} diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseTestCase.java new file mode 100644 index 0000000000..ebf982843c --- /dev/null +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseTestCase.java @@ -0,0 +1,476 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import graphql.ExecutionResult; +import graphql.parser.Parser; +import graphql.schema.idl.SchemaParser; +import graphql.schema.idl.UnExecutableSchemaGenerator; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.Client; +import org.restlet.Component; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Preference; +import org.restlet.data.Protocol; +import org.restlet.data.Reference; +import org.restlet.data.ServerSentEvent; +import org.restlet.data.Status; +import org.restlet.ext.graphql.internal.GraphqlJson; +import org.restlet.representation.EventStreamReader; +import org.restlet.representation.EventStreamRepresentation; +import org.restlet.representation.StringRepresentation; + +/** Subscriptions over {@code graphql-sse}, on the server and the client. */ +class GraphqlSseTestCase { + + private Component component; + + private Client client; + + private String base; + + private GraphqlServerAdapter sse; + + private GraphqlServerAdapter single; + + @BeforeEach + void setUp() throws Exception { + component = new Component(); + component.getServers().add(Protocol.HTTP, 0); + + sse = adapter(); + sse.setGetEnabled(true); + sse.setHeartbeatInterval(Duration.ofMillis(50)); + component.getDefaultHost().attach("/graphql", sse); + + single = adapter(); + single.setMaxSubscriptions(1); + single.setMaxQueuedBytes(16); + single.setHeartbeatInterval(null); + component.getDefaultHost().attach("/limited", single); + + GraphqlServerAdapter disabled = new GraphqlServerAdapter(); + disabled.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + component.getDefaultHost().attach("/disabled", disabled); + + component + .getDefaultHost() + .attach( + "/truncated", + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new EventStreamRepresentation( + sink -> + sink.send( + ServerSentEvent.builder() + .event("next") + .data("{\"data\":{}}") + .build()))); + } + }); + component + .getDefaultHost() + .attach( + "/html", + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setStatus(Status.SERVER_ERROR_BAD_GATEWAY); + response.setEntity("Bad gateway", MediaType.TEXT_HTML); + } + }); + component.start(); + base = "http://localhost:" + component.getServers().get(0).getActualPort(); + client = new Client(Protocol.HTTP); + client.start(); + } + + @AfterEach + void tearDown() throws Exception { + client.stop(); + component.stop(); + } + + private static GraphqlServerAdapter adapter() { + GraphqlServerAdapter adapter = new GraphqlServerAdapter(); + adapter.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + adapter.getSubscriptionTransports().add(GraphqlTransport.SSE); + return adapter; + } + + private Response stream(String path, String query) { + Request request = new Request(Method.POST, base + path); + request.setEntity( + new StringRepresentation( + GraphqlJson.writeString(Map.of("query", query)), + MediaType.APPLICATION_JSON)); + request.getClientInfo() + .getAcceptedMediaTypes() + .add(new Preference<>(MediaType.TEXT_EVENT_STREAM)); + return client.handle(request); + } + + private GraphqlClientAdapter graphql(String path) { + GraphqlClientAdapter graphql = new GraphqlClientAdapter(); + graphql.setClient(client); + graphql.setEndpoint(base + path); + return graphql; + } + + private static void await(BooleanSupplier condition) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (!condition.getAsBoolean()) { + assertTrue(System.nanoTime() < deadline, "Timed out"); + Thread.sleep(10); + } + } + + /** Reads the next event, skipping comments. */ + private static ServerSentEvent nextEvent(EventStreamReader reader) throws Exception { + ServerSentEvent event; + do { + event = reader.readEvent(); + } while (event != null && event.getEvent() == null); + return event; + } + + private static List events(Response response) throws Exception { + List events = new ArrayList<>(); + try (EventStreamReader reader = new EventStreamReader(response.getEntity().getStream())) { + ServerSentEvent event; + while ((event = reader.readEvent()) != null) { + if (event.getEvent() != null) { + events.add(event); + } + } + } + return events; + } + + @Test + void subscriptionStreamsNextThenComplete() throws Exception { + Response response = stream("/graphql", "subscription { ticks(count: 3) }"); + assertEquals(Status.SUCCESS_OK, response.getStatus()); + assertTrue(MediaType.TEXT_EVENT_STREAM.equals(response.getEntity().getMediaType(), true)); + List events = events(response); + assertEquals(4, events.size()); + for (int i = 0; i < 3; i++) { + assertEquals("next", events.get(i).getEvent()); + assertEquals("{\"data\":{\"ticks\":" + (i + 1) + "}}", events.get(i).getData()); + } + assertEquals("complete", events.get(3).getEvent()); + assertEquals("", events.get(3).getData()); + await(() -> sse.getActiveSubscriptions() == 0); + } + + @Test + void queriesOverSseAreOneNextThenComplete() throws Exception { + List events = events(stream("/graphql", "{ hello }")); + assertEquals(2, events.size()); + assertEquals("{\"data\":{\"hello\":\"Hello world\"}}", events.get(0).getData()); + assertEquals("complete", events.get(1).getEvent()); + } + + @Test + void requestErrorsAreJson() throws Exception { + Response response = stream("/graphql", "subscription { unknown }"); + assertEquals(Status.CLIENT_ERROR_BAD_REQUEST, response.getStatus()); + assertTrue(MediaType.APPLICATION_JSON.equals(response.getEntity().getMediaType(), true)); + assertTrue(response.getEntity().getText().contains("\"errors\"")); + } + + @Test + void getAcceptsSubscriptionsOverSse() throws Exception { + Reference reference = new Reference(base + "/graphql"); + reference.addQueryParameter("query", "subscription { ticks(count: 1) }"); + Request request = new Request(Method.GET, reference); + request.getClientInfo() + .getAcceptedMediaTypes() + .add(new Preference<>(MediaType.TEXT_EVENT_STREAM)); + assertEquals(2, events(client.handle(request)).size()); + + Reference mutation = new Reference(base + "/graphql"); + mutation.addQueryParameter("query", "mutation { increment(by: 1) }"); + Request mutationRequest = new Request(Method.GET, mutation); + mutationRequest + .getClientInfo() + .getAcceptedMediaTypes() + .add(new Preference<>(MediaType.TEXT_EVENT_STREAM)); + assertEquals( + Status.CLIENT_ERROR_METHOD_NOT_ALLOWED, client.handle(mutationRequest).getStatus()); + } + + @Test + void sseIsOffUnlessEnabled() { + assertEquals( + Status.CLIENT_ERROR_NOT_ACCEPTABLE, + stream("/disabled", "subscription { ticks(count: 1) }").getStatus()); + } + + @Test + void clientDisconnectCancelsTheSubscription() throws Exception { + int cancelled = TickPublisher.CANCELLED.get(); + Response response = stream("/graphql", "subscription { ticks(count: -1) }"); + try (EventStreamReader reader = new EventStreamReader(response.getEntity().getStream())) { + assertEquals("next", nextEvent(reader).getEvent()); + assertEquals("next", nextEvent(reader).getEvent()); + } + response.abort(); + await(() -> TickPublisher.CANCELLED.get() > cancelled); + await(() -> sse.getActiveSubscriptions() == 0); + } + + @Test + void heartbeatsDetectDisconnectedIdleClients() throws Exception { + int cancelled = TickPublisher.CANCELLED.get(); + Response response = stream("/graphql", "subscription { idle }"); + try (EventStreamReader reader = new EventStreamReader(response.getEntity().getStream())) { + ServerSentEvent heartbeat = reader.readEvent(); + assertNull(heartbeat.getEvent()); + assertEquals("", heartbeat.getComment()); + assertEquals(1, sse.getActiveSubscriptions()); + } + response.abort(); + await(() -> TickPublisher.CANCELLED.get() > cancelled); + await(() -> sse.getActiveSubscriptions() == 0); + } + + @Test + void subscriptionsAreCapped() throws Exception { + Response first = stream("/limited", "subscription { idle }"); + await(() -> single.getActiveSubscriptions() == 1); + Response second = stream("/limited", "subscription { idle }"); + assertEquals(Status.SERVER_ERROR_SERVICE_UNAVAILABLE, second.getStatus()); + first.abort(); + } + + @Test + void queuedBytesAreBounded() throws Exception { + int cancelled = TickPublisher.CANCELLED.get(); + List events = + events(stream("/limited", "subscription { ticks(count: 3) }")); + assertEquals(2, events.size()); + assertTrue(events.get(0).getData().contains("queued bytes limit")); + assertEquals("complete", events.get(1).getEvent()); + assertTrue(TickPublisher.CANCELLED.get() > cancelled); + } + + @Test + void clientHonorsDemandAndCompletes() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { ticks(count: 3) }", null, null).subscribe(subscriber); + + subscriber.subscription.request(1); + assertEquals(Map.of("ticks", 1), subscriber.next().getData()); + Thread.sleep(100); + assertTrue(subscriber.items.isEmpty(), "No result without demand"); + + subscriber.subscription.request(5); + assertEquals(Map.of("ticks", 2), subscriber.next().getData()); + assertEquals(Map.of("ticks", 3), subscriber.next().getData()); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertNull(subscriber.error.get()); + } + + @Test + void clientCancellationAbortsTheRequest() throws Exception { + int cancelled = TickPublisher.CANCELLED.get(); + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe(Parser.parse("subscription { ticks(count: -1) }"), null, Map.of()) + .subscribe(subscriber); + subscriber.subscription.request(2); + subscriber.next(); + subscriber.next(); + subscriber.subscription.cancel(); + await(() -> TickPublisher.CANCELLED.get() > cancelled); + await(() -> sse.getActiveSubscriptions() == 0); + assertEquals(1, subscriber.done.getCount()); + } + + @Test + void clientStopFailsOpenSubscriptions() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber running = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(running); + running.subscription.request(1); + await(() -> sse.getActiveSubscriptions() == 1); + TestSubscriber waiting = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(waiting); + + graphql.stop(); + assertTrue(running.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, running.error.get()); + assertTrue(waiting.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, waiting.error.get()); + assertTrue(client.isStarted()); + await(() -> sse.getActiveSubscriptions() == 0); + assertThrows( + IllegalStateException.class, + () -> graphql.subscribe("subscription { idle }", null, null)); + } + + @Test + void clientPublishesRequestErrorsAsResults() throws Exception { + GraphqlClientAdapter remote = graphql("/graphql"); + remote.start(); + TestSubscriber subscriber = new TestSubscriber(); + remote.subscribe("subscription { unknown }", null, null).subscribe(subscriber); + subscriber.subscription.request(1); + assertFalse(subscriber.next().isDataPresent()); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertNull(subscriber.error.get()); + + GraphqlClientAdapter local = graphql("/graphql"); + local.setSchema( + UnExecutableSchemaGenerator.makeUnExecutableSchema( + new SchemaParser().parse(GraphqlTestSchemas.SDL))); + local.start(); + for (Publisher publisher : + List.of( + local.subscribe("subscription { unknown }", null, null), + local.subscribe("subscription {", null, null), + local.subscribe(Parser.parse("subscription { unknown }"), null, null))) { + TestSubscriber invalid = new TestSubscriber(); + publisher.subscribe(invalid); + invalid.subscription.request(1); + assertFalse(invalid.next().isDataPresent()); + assertTrue(invalid.done.await(5, TimeUnit.SECONDS)); + } + TestSubscriber valid = new TestSubscriber(); + local.subscribe("subscription { ticks(count: 1) }", null, null).subscribe(valid); + valid.subscription.request(Long.MAX_VALUE); + assertEquals(Map.of("ticks", 1), valid.next().getData()); + } + + @Test + void clientSignalsTransportErrors() throws Exception { + for (String path : List.of("/html", "/truncated")) { + GraphqlClientAdapter graphql = graphql(path); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { ticks(count: 1) }", null, null).subscribe(subscriber); + subscriber.subscription.request(Long.MAX_VALUE); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS), path); + assertInstanceOf(GraphqlTransportException.class, subscriber.error.get(), path); + } + + GraphqlClientAdapter limited = graphql("/graphql"); + limited.setMaxResponseBytes(1024); + limited.start(); + TestSubscriber large = new TestSubscriber(); + limited.subscribe( + "query($n: String) { hello(name: $n) }", + null, + Map.of("n", "x".repeat(20_000))) + .subscribe(large); + large.subscription.request(Long.MAX_VALUE); + assertTrue(large.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, large.error.get()); + } + + @Test + void clientRejectsNonPositiveDemand() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { ticks(count: 1) }", null, null).subscribe(subscriber); + subscriber.subscription.request(0); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, subscriber.error.get()); + } + + @Test + void configurationIsValidated() { + GraphqlServerAdapter adapter = new GraphqlServerAdapter(); + assertThrows(IllegalArgumentException.class, () -> adapter.setMaxSubscriptions(0)); + assertThrows(IllegalArgumentException.class, () -> adapter.setMaxQueuedBytes(0)); + assertThrows( + IllegalArgumentException.class, () -> adapter.setHeartbeatInterval(Duration.ZERO)); + assertEquals(GraphqlServerAdapter.DEFAULT_MAX_SUBSCRIPTIONS, adapter.getMaxSubscriptions()); + assertEquals(GraphqlServerAdapter.DEFAULT_MAX_QUEUED_BYTES, adapter.getMaxQueuedBytes()); + assertEquals( + GraphqlServerAdapter.DEFAULT_HEARTBEAT_INTERVAL, adapter.getHeartbeatInterval()); + assertThrows( + UnsupportedOperationException.class, + () -> sse.getSubscriptionTransports().remove(GraphqlTransport.SSE)); + } + + /** Collects the signals of a subscription. */ + private static final class TestSubscriber implements Subscriber { + + final BlockingQueue items = new LinkedBlockingQueue<>(); + + final CountDownLatch done = new CountDownLatch(1); + + final AtomicReference error = new AtomicReference<>(); + + volatile Subscription subscription; + + @Override + public void onSubscribe(Subscription subscription) { + this.subscription = subscription; + } + + @Override + public void onNext(ExecutionResult item) { + items.add(item); + } + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + done.countDown(); + } + + @Override + public void onComplete() { + done.countDown(); + } + + ExecutionResult next() throws InterruptedException { + ExecutionResult item = items.poll(5, TimeUnit.SECONDS); + assertTrue(item != null, "Timed out waiting for a result"); + return item; + } + } +} diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlTestSchemas.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlTestSchemas.java index c08d563360..9b74cae46d 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlTestSchemas.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlTestSchemas.java @@ -47,6 +47,7 @@ final class GraphqlTestSchemas { } type Subscription { ticks(count: Int!): Int + idle: Int } """; @@ -114,7 +115,12 @@ static GraphQLSchema schema(AtomicInteger counter) { type.dataFetcher( "increment", env -> counter.addAndGet(env.getArgument("by")))) - .type("Subscription", type -> type.dataFetcher("ticks", ticks)) + .type( + "Subscription", + type -> + type.dataFetcher("ticks", ticks) + .dataFetcher( + "idle", env -> new TickPublisher(0, true))) .build(); return new SchemaGenerator().makeExecutableSchema(new SchemaParser().parse(SDL), wiring); } diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java index 5d7f308eba..3e4384f924 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java @@ -15,7 +15,8 @@ /** * Publishes 1, 2, ... up to a count, or forever when the count is negative, honoring demand and - * cancellation. Counts its instances and cancellations for the tests. + * cancellation. An idle publisher never publishes. Counts its instances and cancellations for the + * tests. */ final class TickPublisher implements Publisher { @@ -25,8 +26,15 @@ final class TickPublisher implements Publisher { private final int count; + private final boolean idle; + TickPublisher(int count) { + this(count, false); + } + + TickPublisher(int count, boolean idle) { this.count = count; + this.idle = idle; CREATED.incrementAndGet(); } @@ -53,6 +61,9 @@ private final class Ticks implements Subscription { @Override public void request(long n) { + if (idle) { + return; + } synchronized (this) { demand += n; if (emitting) { From 08f497ffb40ed610fdec16edf6b6d71e0caa3a43 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:13:32 -0400 Subject: [PATCH 2/2] fix: harden the GraphQL SSE subscription lifecycle On the server, release the subscription slot when the first write fails, answer document errors of event stream requests with 400 and keep adapters without subscription transports restartable. On the client, signal non-positive demand without sending or waiting for a request, unregister subscriptions cancelled from onSubscribe() and abort requests cancelled before their exchange started. Co-Authored-By: Claude Opus 5.5 --- .../ext/graphql/GraphqlClientAdapter.java | 15 +- .../ext/graphql/GraphqlServerAdapter.java | 10 +- .../ext/graphql/GraphqlSsePublisher.java | 67 ++++++-- .../graphql/internal/GraphqlEventStream.java | 4 +- .../graphql/GraphqlSseEdgeCasesTestCase.java | 152 ++++++++++++++++++ 5 files changed, 228 insertions(+), 20 deletions(-) create mode 100644 org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseEdgeCasesTestCase.java diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java index 7b1116b1df..aaa7928aab 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java @@ -441,16 +441,17 @@ void unregister(GraphqlSsePublisher.EventStreamSubscription subscription) { } /** - * Creates the HTTP request of an operation sent for an event stream. + * 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( - String document, String operationName, Map variables) { - Request request = createRequest(document, operationName, variables); + Request request, String document, String operationName, Map variables) { + createRequest(request, document, operationName, variables); request.getClientInfo() .getAcceptedMediaTypes() .add(new Preference<>(MediaType.TEXT_EVENT_STREAM, 1F)); @@ -491,6 +492,11 @@ private static ExecutionResult failure(List errors) { * @return The request, customized. */ Request createRequest(String document, String operationName, Map variables) { + return createRequest(new Request(), document, operationName, variables); + } + + private Request createRequest( + Request request, String document, String operationName, Map variables) { Map body = new LinkedHashMap<>(); body.put("query", document); if (operationName != null) { @@ -500,7 +506,8 @@ Request createRequest(String document, String operationName, Map 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), diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java index 04b3894af9..e92af015f2 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java @@ -353,8 +353,10 @@ public synchronized void start() throws Exception { if (parserOptions == null) { parserOptions = ParserOptions.getDefaultOperationParserOptions(); } - subscriptionTransports = - Collections.unmodifiableSet(EnumSet.copyOf(subscriptionTransports)); + // A typed copy, as EnumSet.copyOf() fails on an empty set once it is wrapped. + EnumSet transports = EnumSet.noneOf(GraphqlTransport.class); + transports.addAll(subscriptionTransports); + subscriptionTransports = Collections.unmodifiableSet(transports); super.start(); } } @@ -437,10 +439,10 @@ public void handle(Request request, Response response) { operation = selectOperation(executionInput.getQuery(), executionInput.getOperationName()); } catch (InvalidSyntaxException e) { - write(response, responseType, failure(e.toInvalidSyntaxError())); + write(response, responseType, failure(e.toInvalidSyntaxError()), sse); return; } catch (RequestException e) { - write(response, responseType, failure(error(e.getMessage()))); + write(response, responseType, failure(error(e.getMessage())), sse); return; } diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java index 319344d85f..ae34de521c 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java @@ -14,6 +14,7 @@ import java.io.InputStream; import java.util.Map; import java.util.Objects; +import java.util.function.BooleanSupplier; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; @@ -58,6 +59,9 @@ public void subscribe(Subscriber subscriber) { subscriber.onSubscribe(subscription); if (!adapter.register(subscription)) { subscription.stop(); + } else if (subscription.isDone()) { + // Cancelled or finished from onSubscribe(), before being registered. + adapter.unregister(subscription); } } @@ -70,10 +74,12 @@ final class EventStreamSubscription implements Subscription { private boolean started; - private boolean invalidDemand; + private volatile boolean invalidDemand; private volatile boolean cancelled; + private volatile boolean finished; + private volatile boolean stopped; private volatile Request request; @@ -84,22 +90,43 @@ final class EventStreamSubscription implements Subscription { @Override public void request(long n) { + boolean signal = false; + boolean start = false; synchronized (this) { if (cancelled) { return; } if (n <= 0) { invalidDemand = true; + // Without a reader thread, signal from here and prevent one from starting. + signal = !started; + if (signal) { + cancelled = true; + } } else { demand = demand + n < 0 ? Long.MAX_VALUE : demand + n; + start = !started; } + started = true; notifyAll(); - if (started) { - return; + } + if (signal) { + subscriber.onError(invalidDemandError()); + adapter.unregister(this); + } else if (start) { + Thread.ofVirtual().name("restlet-graphql-sse").start(this::run); + } else if (n <= 0) { + // Unblocks a pending read, then the reader thread signals the error. + Request current = request; + if (current != null) { + current.abort(); } - started = true; } - Thread.ofVirtual().name("restlet-graphql-sse").start(this::run); + } + + /** Indicates if the subscription is cancelled or its reader thread is done. */ + boolean isDone() { + return cancelled || finished; } @Override @@ -136,6 +163,10 @@ void stop() { cancel(); } + private IllegalArgumentException invalidDemandError() { + return new IllegalArgumentException("The requested number of results must be positive"); + } + private GraphqlTransportException stoppedError() { return new GraphqlTransportException( "The GraphQL client was stopped", null, null, null); @@ -145,7 +176,8 @@ private void run() { Response response = null; try { Request current = - adapter.createEventStreamRequest(document, operationName, variables); + adapter.createEventStreamRequest( + new EventStreamRequest(), document, operationName, variables); request = current; if (cancelled) { terminateIfStopped(); @@ -184,6 +216,7 @@ private void run() { if (response != null && response.getEntity() != null) { response.getEntity().release(); } + finished = true; adapter.unregister(this); } } @@ -243,9 +276,7 @@ private boolean awaitDemand() { } if (invalidDemand) { cancelled = true; - subscriber.onError( - new IllegalArgumentException( - "The requested number of results must be positive")); + subscriber.onError(invalidDemandError()); return false; } if (demand != Long.MAX_VALUE) { @@ -266,7 +297,23 @@ private void fail(Throwable error) { subscriber.onError(stoppedError()); } else if (!cancelled) { cancelled = true; - subscriber.onError(error); + // A read aborted by invalid demand reports that demand. + subscriber.onError(invalidDemand ? invalidDemandError() : error); + } + } + + /** + * A request aborted as soon as its connector can abort it, when cancelled before the + * exchange started. + */ + private final class EventStreamRequest extends Request { + + @Override + public void setAbortAction(BooleanSupplier action) { + super.setAbortAction(action); + if (action != null && (cancelled || invalidDemand)) { + action.getAsBoolean(); + } } } } diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java index 4c52d08035..050fb1698e 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java @@ -159,12 +159,12 @@ public void write(EventSink sink) throws IOException { sink.send(COMPLETE); return; } - // An empty comment sends the response head at once, before the first result. - sink.send(OPEN); QueueingSubscriber subscriber = new QueueingSubscriber(maxQueuedBytes); this.subscriber = subscriber; openStreams.add(this); try { + // An empty comment sends the response head at once, before the first result. + sink.send(OPEN); // Registered before checking, so that a concurrent stop of the adapter sees it. if (stopped || endpointStopped.getAsBoolean()) { subscriber.fail(new Failure("The GraphQL endpoint is stopping", null)); diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseEdgeCasesTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseEdgeCasesTestCase.java new file mode 100644 index 0000000000..9803beff4d --- /dev/null +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseEdgeCasesTestCase.java @@ -0,0 +1,152 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import graphql.ExecutionResult; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.Client; +import org.restlet.Component; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Preference; +import org.restlet.data.Protocol; +import org.restlet.data.Status; +import org.restlet.ext.graphql.internal.GraphqlJson; +import org.restlet.representation.StringRepresentation; + +/** Edge cases of the {@code graphql-sse} binding: request errors, restarts and demand. */ +class GraphqlSseEdgeCasesTestCase { + + private final AtomicInteger requests = new AtomicInteger(); + + private Component component; + + private Client client; + + private String base; + + @BeforeEach + void setUp() throws Exception { + component = new Component(); + component.getServers().add(Protocol.HTTP, 0); + GraphqlServerAdapter sse = new GraphqlServerAdapter(); + sse.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + sse.getSubscriptionTransports().add(GraphqlTransport.SSE); + component.getDefaultHost().attach("/graphql", sse); + component + .getDefaultHost() + .attach( + "/counted", + new Restlet() { + @Override + public void handle(Request request, Response response) { + requests.incrementAndGet(); + response.setStatus(Status.CLIENT_ERROR_NOT_FOUND); + } + }); + component.start(); + base = "http://localhost:" + component.getServers().get(0).getActualPort(); + client = new Client(Protocol.HTTP); + client.start(); + } + + @AfterEach + void tearDown() throws Exception { + client.stop(); + component.stop(); + } + + private Response stream(Map body) { + Request request = new Request(Method.POST, base + "/graphql"); + request.setEntity( + new StringRepresentation( + GraphqlJson.writeString(body), MediaType.APPLICATION_JSON)); + request.getClientInfo() + .getAcceptedMediaTypes() + .add(new Preference<>(MediaType.TEXT_EVENT_STREAM)); + return client.handle(request); + } + + @Test + void documentErrorsAreBadRequests() { + assertEquals( + Status.CLIENT_ERROR_BAD_REQUEST, + stream(Map.of("query", "subscription {")).getStatus()); + assertEquals( + Status.CLIENT_ERROR_BAD_REQUEST, + stream(Map.of("query", "{ hello }", "operationName", "Unknown")).getStatus()); + assertEquals( + Status.CLIENT_ERROR_BAD_REQUEST, + stream(Map.of("query", "query A { hello } query B { hello }")).getStatus()); + } + + @Test + void adapterWithoutTransportsRestarts() throws Exception { + GraphqlServerAdapter adapter = new GraphqlServerAdapter(); + adapter.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + adapter.start(); + adapter.stop(); + adapter.start(); + assertTrue(adapter.isStarted()); + assertTrue(adapter.getSubscriptionTransports().isEmpty()); + adapter.stop(); + } + + @Test + void nonPositiveDemandFailsWithoutRequest() throws Exception { + GraphqlClientAdapter graphql = new GraphqlClientAdapter(); + graphql.setClient(client); + graphql.setEndpoint(base + "/counted"); + graphql.start(); + CountDownLatch done = new CountDownLatch(1); + AtomicReference error = new AtomicReference<>(); + graphql.subscribe("subscription { ticks(count: 1) }", null, null) + .subscribe( + new Subscriber() { + @Override + public void onSubscribe(Subscription subscription) { + subscription.request(0); + } + + @Override + public void onNext(ExecutionResult item) {} + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + done.countDown(); + } + + @Override + public void onComplete() { + done.countDown(); + } + }); + assertTrue(done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, error.get()); + assertEquals(0, requests.get()); + graphql.stop(); + } +}