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..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 @@ -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,128 @@ 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); + } + + /** + * 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 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 @@ 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) { @@ -365,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), @@ -401,7 +543,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..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 @@ -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,26 @@ public synchronized void start() throws Exception { if (parserOptions == null) { parserOptions = ParserOptions.getDefaultOperationParserOptions(); } + // 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(); } } + /** + * 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 +389,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; @@ -281,26 +439,56 @@ 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; } - 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 +652,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..ae34de521c --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java @@ -0,0 +1,368 @@ +/** + * 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 java.util.function.BooleanSupplier; +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(); + } else if (subscription.isDone()) { + // Cancelled or finished from onSubscribe(), before being registered. + adapter.unregister(subscription); + } + } + + /** The subscription of one subscriber. */ + final class EventStreamSubscription implements Subscription { + + private final Subscriber subscriber; + + private long demand; + + private boolean started; + + private volatile boolean invalidDemand; + + private volatile boolean cancelled; + + private volatile boolean finished; + + private volatile boolean stopped; + + private volatile Request request; + + EventStreamSubscription(Subscriber subscriber) { + this.subscriber = subscriber; + } + + @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 (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(); + } + } + } + + /** Indicates if the subscription is cancelled or its reader thread is done. */ + boolean isDone() { + return cancelled || finished; + } + + @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 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); + } + + private void run() { + Response response = null; + try { + Request current = + adapter.createEventStreamRequest( + new EventStreamRequest(), 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(); + } + finished = true; + 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(invalidDemandError()); + 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; + // 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(); + } + } + } + } + + /** + * 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..050fb1698e --- /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; + } + 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)); + } 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/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(); + } +} 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) {