From 2d90470baf91f21200a84c1bbe347e090f850e49 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 22:57:08 -0400 Subject: [PATCH 1/3] feat: add the GraphQL WebSocket server (G4, graphql-transport-ws) - GraphqlTransport.WEBSOCKET: the server adapter accepts WebSocket upgrades offering the graphql-transport-ws subprotocol on its own path, after the same filters and authenticators; other subprotocols get 400. HTTP keeps working on the same path. - Protocol: connection_init and connection_ack with a configurable timeout (3 s, close 4408), ping and pong, several operations per connection on their own virtual threads, complete in both directions, error messages for errors before execution, and close codes 4400 (invalid message), 4401 (subscribe before ack), 4409 (duplicate id) and 4429 (repeated connection_init). - The connection_init payload reaches data fetchers under RestletGraphql.CONNECTION_INIT_PAYLOAD. - Subscriptions share the SSE limits and back-pressure: the queueing subscriber of the SSE stream is extracted as GraphqlResultQueue. Closing a connection cancels its subscriptions; stopping the adapter closes connections with 1001. - org.restlet.ext.websocket is optional (requires static, optional Maven dependency) and only loaded when WEBSOCKET is enabled. The named-module test now runs without it. Refs #1586 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 6 +- org.restlet.ext.graphql/pom.xml | 35 ++ .../src/main/java/module-info.java | 3 + .../ext/graphql/GraphqlServerAdapter.java | 86 +++- .../restlet/ext/graphql/GraphqlTransport.java | 9 +- .../ext/graphql/GraphqlUpgradeSupport.java | 32 ++ .../ext/graphql/GraphqlWebSocketSupport.java | 379 ++++++++++++++++++ .../restlet/ext/graphql/RestletGraphql.java | 7 + .../graphql/internal/GraphqlEventStream.java | 128 +----- .../graphql/internal/GraphqlResultQueue.java | 165 ++++++++ .../graphql/GraphqlModulePathTestCase.java | 6 +- .../ext/graphql/GraphqlTestSchemas.java | 10 +- .../GraphqlWebSocketServerTestCase.java | 328 +++++++++++++++ 13 files changed, 1065 insertions(+), 129 deletions(-) create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlUpgradeSupport.java create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java create mode 100644 org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java diff --git a/changes.md b/changes.md index 9baeb50d35..7343d885c4 100644 --- a/changes.md +++ b/changes.md @@ -92,7 +92,11 @@ Changes log result requested at a time, client disconnects cancel the subscription, and finite limits apply (64 active subscriptions and 1 MiB of queued bytes per subscription by default). `GraphqlClientAdapter.subscribe` returns a `Publisher` - honoring demand; cancelling it aborts the request. Issue #1581. + honoring demand; cancelling it aborts the request. With `GraphqlTransport.WEBSOCKET`, + the server adapter also accepts `graphql-transport-ws` WebSocket upgrades on its own path + (HTTP/1.1 only), with several operations per connection, a `connection_init` timeout + and the protocol's close codes; `org.restlet.ext.websocket` is an optional dependency, + loaded only then. Issue #1581. - Added the `NettyConnectionUpgrade` engine hook: a Restlet returning a "101 Switching Protocols" response with this response attribute takes over the HTTP/1.1 connection (`NettyUpgradeContext`), which the WebSocket extension relies on. diff --git a/org.restlet.ext.graphql/pom.xml b/org.restlet.ext.graphql/pom.xml index fbf42d79d1..52c8f5c172 100644 --- a/org.restlet.ext.graphql/pom.xml +++ b/org.restlet.ext.graphql/pom.xml @@ -29,6 +29,13 @@ jackson-databind ${lib-jackson3-version} + + org.restlet + org.restlet.ext.websocket + ${project.version} + + true + org.junit.jupiter junit-jupiter-api @@ -36,4 +43,32 @@ test + + + m2e-test-modules + + m2e.version + + + + + maven-compiler-plugin + + + default-testCompile + + + + --add-reads + org.restlet.ext.graphql=java.net.http,jdk.httpserver + + + + + + + + + diff --git a/org.restlet.ext.graphql/src/main/java/module-info.java b/org.restlet.ext.graphql/src/main/java/module-info.java index 7f6f6e54d2..c18fe5a98d 100644 --- a/org.restlet.ext.graphql/src/main/java/module-info.java +++ b/org.restlet.ext.graphql/src/main/java/module-info.java @@ -16,6 +16,9 @@ requires transitive com.graphqljava; requires transitive org.reactivestreams; + // Optional: only loaded when GraphqlTransport.WEBSOCKET is enabled. + requires static org.restlet.ext.websocket; + // JSON is an implementation detail: required, not re-exported. requires tools.jackson.core; requires tools.jackson.databind; 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 e92af015f2..bed7f4cba5 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 @@ -83,6 +83,12 @@ * {@code complete} event. A client disconnect, detected on write or by heartbeats, cancels the * subscription, and stopping the adapter ends its open streams. * + *

With {@link GraphqlTransport#WEBSOCKET}, the adapter also accepts WebSocket upgrades offering + * the {@code graphql-transport-ws} subprotocol on its own path, on HTTP/1.1 connections. The + * upgrade request goes through the same filters and authenticators, and the payload of the {@code + * connection_init} message is available under {@link RestletGraphql#CONNECTION_INIT_PAYLOAD}. + * Subscriptions share the limits of SSE ones; stopping the adapter closes the connections. + * *

Operations run on the thread of the call, a virtual thread by default. Asynchronous data * fetchers are joined on that thread. The {@code graphql.GraphQLContext} of each operation holds * the Restlet request under {@link RestletGraphql#REQUEST} and its client info under {@link @@ -104,6 +110,9 @@ public class GraphqlServerAdapter extends Restlet { /** The default maximum number of bytes queued for a subscription: 1 MiB. */ public static final long DEFAULT_MAX_QUEUED_BYTES = 1L << 20; + /** The default time allowed to WebSocket clients to initialize their connection: 3 seconds. */ + public static final Duration DEFAULT_CONNECTION_INIT_TIMEOUT = Duration.ofSeconds(3); + /** The default interval of event stream heartbeats: 15 seconds. */ public static final Duration DEFAULT_HEARTBEAT_INTERVAL = Duration.ofSeconds(15); @@ -130,6 +139,10 @@ public class GraphqlServerAdapter extends Restlet { private final Set openStreams = ConcurrentHashMap.newKeySet(); + private volatile Duration connectionInitTimeout = DEFAULT_CONNECTION_INIT_TIMEOUT; + + private volatile GraphqlUpgradeSupport upgradeSupport; + /** Constructor. */ public GraphqlServerAdapter() { this(null); @@ -335,6 +348,37 @@ public void setHeartbeatInterval(Duration heartbeatInterval) { this.heartbeatInterval = heartbeatInterval; } + /** + * Returns the time allowed to WebSocket clients to send their {@code connection_init} message. + * + * @return The time allowed to initialize a WebSocket connection. + */ + public Duration getConnectionInitTimeout() { + return connectionInitTimeout; + } + + /** + * Sets the time allowed to WebSocket clients to send their {@code connection_init} message, + * after which the connection is closed with code 4408. Defaults to {@link + * #DEFAULT_CONNECTION_INIT_TIMEOUT}. + * + * @param connectionInitTimeout The time allowed to initialize a WebSocket connection. + */ + public void setConnectionInitTimeout(Duration connectionInitTimeout) { + checkNotStarted(); + if (connectionInitTimeout == null + || connectionInitTimeout.isZero() + || connectionInitTimeout.isNegative()) { + throw new IllegalArgumentException("The connection init timeout must be positive"); + } + this.connectionInitTimeout = connectionInitTimeout; + } + + /** Returns the counter of active subscriptions, shared by all transports. */ + AtomicInteger getActiveSubscriptionCounter() { + return activeSubscriptions; + } + /** * Returns the number of active subscriptions. * @@ -357,6 +401,10 @@ public synchronized void start() throws Exception { EnumSet transports = EnumSet.noneOf(GraphqlTransport.class); transports.addAll(subscriptionTransports); subscriptionTransports = Collections.unmodifiableSet(transports); + if (subscriptionTransports.contains(GraphqlTransport.WEBSOCKET)) { + // Only loads the optional WebSocket extension when enabled. + upgradeSupport = new GraphqlWebSocketSupport(this); + } super.start(); } } @@ -371,6 +419,10 @@ public synchronized void stop() throws Exception { for (GraphqlEventStream stream : List.copyOf(openStreams)) { stream.stop(); } + GraphqlUpgradeSupport upgrades = upgradeSupport; + if (upgrades != null) { + upgrades.stop(); + } } @Override @@ -379,6 +431,10 @@ public void handle(Request request, Response response) { if (!isStarted()) { return; } + GraphqlUpgradeSupport upgrades = upgradeSupport; + if (upgrades != null && upgrades.handle(request, response)) { + return; + } Method method = request.getMethod(); boolean get = Method.GET.equals(method); @@ -433,7 +489,7 @@ public void handle(Request request, Response response) { } // The operation is selected once customized, as the customizer can change it. - ExecutionInput executionInput = customize(request, input).build(); + ExecutionInput executionInput = customize(request, input, null).build(); OperationDefinition.Operation operation; try { operation = @@ -491,17 +547,35 @@ public void handle(Request request, Response response) { } } - private ExecutionInput.Builder customize(Request request, ExecutionInput.Builder input) { + /** + * Executes an operation. + * + * @param request The request. + * @param input The execution input, before customization. + * @param initPayload The payload of a WebSocket connection_init message, or null. + * @return The result. + */ + ExecutionResult execute( + Request request, ExecutionInput.Builder input, Map initPayload) { + return graphQL.execute(customize(request, input, initPayload).build()); + } + + private ExecutionInput.Builder customize( + Request request, ExecutionInput.Builder input, Map initPayload) { BiConsumer customizer = executionInputCustomizer; if (customizer != null) { customizer.accept(request, input); } // graphql-java reads per-execution parser options from the context under their class. return input.graphQLContext( - builder -> - builder.put(ParserOptions.class, parserOptions) - .put(RestletGraphql.REQUEST, request) - .put(RestletGraphql.CLIENT_INFO, request.getClientInfo())); + builder -> { + builder.put(ParserOptions.class, parserOptions) + .put(RestletGraphql.REQUEST, request) + .put(RestletGraphql.CLIENT_INFO, request.getClientInfo()); + if (initPayload != null) { + builder.put(RestletGraphql.CONNECTION_INIT_PAYLOAD, initPayload); + } + }); } /** 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 index 3f4111c3da..9b91a9fbf7 100644 --- 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 @@ -19,5 +19,12 @@ public enum GraphqlTransport { * a request accepting {@code text/event-stream} gets one {@code next} event per result and a * final {@code complete} event. */ - SSE + SSE, + + /** + * WebSocket, following the {@code graphql-transport-ws} protocol: the adapter accepts upgrades + * offering this subprotocol, on HTTP/1.1 connections only. Requires the {@code + * org.restlet.ext.websocket} module. + */ + WEBSOCKET } diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlUpgradeSupport.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlUpgradeSupport.java new file mode 100644 index 0000000000..5a9c3770d9 --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlUpgradeSupport.java @@ -0,0 +1,32 @@ +/** + * 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 org.restlet.Request; +import org.restlet.Response; + +/** + * Handles connection upgrades for {@link GraphqlServerAdapter}. The implementation depends on the + * optional WebSocket extension: the adapter only refers to it through this interface, so that the + * extension is loaded only when {@link GraphqlTransport#WEBSOCKET} is enabled. + */ +interface GraphqlUpgradeSupport { + + /** + * Handles a request if it asks for a connection upgrade. + * + * @param request The request. + * @param response The response. + * @return True if the request asked for an upgrade and was handled. + */ + boolean handle(Request request, Response response); + + /** Closes the open connections. */ + void stop(); +} diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java new file mode 100644 index 0000000000..c193f8903b --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java @@ -0,0 +1,379 @@ +/** + * 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.ExecutionInput; +import graphql.ExecutionResult; +import graphql.GraphQLError; +import graphql.GraphqlErrorBuilder; +import java.io.IOException; +import java.util.LinkedHashMap; +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.logging.Level; +import org.reactivestreams.Publisher; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.Status; +import org.restlet.ext.graphql.internal.GraphqlJson; +import org.restlet.ext.graphql.internal.GraphqlResultQueue; +import org.restlet.ext.websocket.WebSocketMessage; +import org.restlet.ext.websocket.WebSocketSession; +import org.restlet.ext.websocket.WebSocketUpgrade; + +/** + * Serves the {@code graphql-transport-ws} protocol on WebSocket upgrades of a {@link + * GraphqlServerAdapter}. Each connection runs on the virtual thread of its session; each operation + * runs on its own virtual thread. + */ +final class GraphqlWebSocketSupport implements GraphqlUpgradeSupport { + + /** The WebSocket subprotocol. */ + static final String SUBPROTOCOL = "graphql-transport-ws"; + + // Close codes of graphql-transport-ws. + static final int INVALID_MESSAGE = 4400; + + static final int UNAUTHORIZED = 4401; + + static final int INIT_TIMEOUT = 4408; + + static final int SUBSCRIBER_EXISTS = 4409; + + static final int TOO_MANY_INIT = 4429; + + private final GraphqlServerAdapter adapter; + + private final Set connections = ConcurrentHashMap.newKeySet(); + + GraphqlWebSocketSupport(GraphqlServerAdapter adapter) { + this.adapter = adapter; + } + + @Override + public boolean handle(Request request, Response response) { + if (!WebSocketUpgrade.isRequested(request)) { + return false; + } + if (!WebSocketUpgrade.getRequestedSubprotocols(request).contains(SUBPROTOCOL)) { + response.setStatus( + Status.CLIENT_ERROR_BAD_REQUEST, + "The " + SUBPROTOCOL + " subprotocol is required"); + return true; + } + WebSocketUpgrade.accept( + response, session -> new Connection(request, session).run(), SUBPROTOCOL); + return true; + } + + @Override + public void stop() { + for (Connection connection : List.copyOf(connections)) { + connection.close(WebSocketSession.GOING_AWAY, "The GraphQL endpoint is stopping"); + } + } + + private static String message(String type, String id, Object payload) { + Map message = new LinkedHashMap<>(); + if (id != null) { + message.put("id", id); + } + message.put("type", type); + if (payload != null) { + message.put("payload", payload); + } + return GraphqlJson.writeString(message); + } + + private static List> errors(List errors) { + return errors.stream().map(GraphQLError::toSpecification).toList(); + } + + private static List> error(String message) { + return errors(List.of(GraphqlErrorBuilder.newError().message("%s", message).build())); + } + + /** A WebSocket connection. */ + private final class Connection { + + private final Request request; + + private final WebSocketSession session; + + private final Map operations = new ConcurrentHashMap<>(); + + private final Object sendLock = new Object(); + + private volatile boolean initialized; + + private volatile Map initPayload; + + Connection(Request request, WebSocketSession session) { + this.request = request; + this.session = session; + } + + void run() throws IOException { + connections.add(this); + Thread timeout = + Thread.ofVirtual().name("restlet-graphql-ws-init").start(this::awaitInit); + try { + WebSocketMessage received; + while ((received = session.receive()) != null) { + if (!received.isText()) { + close(INVALID_MESSAGE, "Binary messages are not supported"); + return; + } + if (!dispatch(GraphqlJson.asObject(parse(received.getText())))) { + return; + } + } + } finally { + timeout.interrupt(); + for (Operation operation : operations.values()) { + operation.cancel(); + } + connections.remove(this); + } + } + + private Object parse(String text) { + try { + return GraphqlJson.parse(text); + } catch (GraphqlJson.InvalidJsonException _) { + return null; + } + } + + private void awaitInit() { + try { + Thread.sleep(adapter.getConnectionInitTimeout()); + } catch (InterruptedException _) { + return; + } + if (!initialized) { + close(INIT_TIMEOUT, "Connection initialisation timeout"); + } + } + + /** Handles a message. Returns false once the connection is closed. */ + private boolean dispatch(Map message) { + Object type = message == null ? null : message.get("type"); + if (!(type instanceof String)) { + return close(INVALID_MESSAGE, "Invalid message"); + } + switch ((String) type) { + case "connection_init" -> { + if (initialized) { + return close(TOO_MANY_INIT, "Too many initialisation requests"); + } + Object payload = message.get("payload"); + if (payload != null && GraphqlJson.asObject(payload) == null) { + return close(INVALID_MESSAGE, "Invalid connection_init payload"); + } + initPayload = GraphqlJson.asObject(payload); + initialized = true; + return send(message("connection_ack", null, null)); + } + case "ping" -> { + return send(message("pong", null, null)); + } + case "pong" -> { + return true; + } + case "subscribe" -> { + return subscribe(message); + } + case "complete" -> { + if (!(message.get("id") instanceof String id)) { + return close(INVALID_MESSAGE, "Invalid complete message"); + } + Operation operation = operations.remove(id); + if (operation != null) { + operation.cancel(); + } + return true; + } + default -> { + return close(INVALID_MESSAGE, "Unknown message type '" + type + "'"); + } + } + } + + private boolean subscribe(Map message) { + if (!initialized) { + return close(UNAUTHORIZED, "Unauthorized"); + } + Map payload = GraphqlJson.asObject(message.get("payload")); + if (!(message.get("id") instanceof String id) + || payload == null + || !(payload.get("query") instanceof String query) + || (payload.get("operationName") != null + && !(payload.get("operationName") instanceof String)) + || invalidObject(payload, "variables") + || invalidObject(payload, "extensions")) { + return close(INVALID_MESSAGE, "Invalid subscribe message"); + } + + ExecutionInput.Builder input = + ExecutionInput.newExecutionInput() + .query(query) + .operationName((String) payload.get("operationName")); + Map variables = GraphqlJson.asObject(payload.get("variables")); + if (variables != null) { + input.variables(variables); + } + Map extensions = GraphqlJson.asObject(payload.get("extensions")); + if (extensions != null) { + input.extensions(extensions); + } + + Operation operation = new Operation(this, id, input); + if (operations.putIfAbsent(id, operation) != null) { + return close(SUBSCRIBER_EXISTS, "Subscriber for " + id + " already exists"); + } + Thread.ofVirtual().name("restlet-graphql-ws").start(operation); + return true; + } + + private boolean invalidObject(Map payload, String name) { + Object value = payload.get(name); + return value != null && GraphqlJson.asObject(value) == null; + } + + /** Sends a message. Returns false if the session is closed. */ + boolean send(String text) { + synchronized (sendLock) { + try { + session.sendText(text); + return true; + } catch (IOException _) { + return false; + } + } + } + + /** Closes the connection. Returns false, as a convenience for the dispatch loop. */ + boolean close(int code, String reason) { + try { + session.close(code, reason); + } catch (IOException e) { + adapter.getLogger().log(Level.FINE, "Unable to close a GraphQL WebSocket", e); + } + return false; + } + } + + /** An operation of a connection. */ + private final class Operation implements Runnable { + + private final Connection connection; + + private final String id; + + private final ExecutionInput.Builder input; + + private volatile GraphqlResultQueue queue; + + private volatile boolean cancelled; + + Operation(Connection connection, String id, ExecutionInput.Builder input) { + this.connection = connection; + this.id = id; + this.input = input; + } + + void cancel() { + cancelled = true; + GraphqlResultQueue current = queue; + if (current != null) { + current.cancelByClient(); + } + } + + @Override + public void run() { + try { + ExecutionResult result = + adapter.execute(connection.request, input, connection.initPayload); + if (cancelled) { + return; + } + if (!result.isDataPresent()) { + connection.send(message("error", id, errors(result.getErrors()))); + } else if (result.getData() instanceof Publisher publisher) { + @SuppressWarnings("unchecked") + Publisher results = (Publisher) publisher; + stream(results); + } else if (connection.send(message("next", id, result.toSpecification()))) { + connection.send(message("complete", id, null)); + } + } catch (RuntimeException e) { + adapter.getLogger().log(Level.WARNING, "GraphQL operation failed", e); + connection.send(message("error", id, error("The operation failed"))); + } finally { + connection.operations.remove(id, this); + } + } + + private void stream(Publisher results) { + AtomicInteger active = adapter.getActiveSubscriptionCounter(); + if (active.incrementAndGet() > adapter.getMaxSubscriptions()) { + active.decrementAndGet(); + connection.send(message("error", id, error("Too many active subscriptions"))); + return; + } + GraphqlResultQueue current = new GraphqlResultQueue(adapter.getMaxQueuedBytes()); + queue = current; + try { + if (cancelled) { + return; + } + results.subscribe(current); + while (true) { + GraphqlResultQueue.Signal signal = current.take(); + if (signal instanceof GraphqlResultQueue.Next next) { + if (!connection.send( + "{\"id\":" + + GraphqlJson.writeString(id) + + ",\"type\":\"next\",\"payload\":" + + next.json() + + "}")) { + return; + } + current.request(); + } else if (signal instanceof GraphqlResultQueue.Failure failure) { + if (failure.cause() != null) { + adapter.getLogger() + .log( + Level.WARNING, + "GraphQL subscription failed", + failure.cause()); + } + connection.send(message("error", id, error(failure.message()))); + return; + } else if (signal instanceof GraphqlResultQueue.Completed) { + connection.send(message("complete", id, null)); + return; + } else { + return; + } + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + } finally { + current.cancel(); + active.decrementAndGet(); + } + } + } +} diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/RestletGraphql.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/RestletGraphql.java index b1f8e5fc0e..73946cd470 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/RestletGraphql.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/RestletGraphql.java @@ -26,5 +26,12 @@ public final class RestletGraphql { */ public static final String CLIENT_INFO = "org.restlet.ext.graphql.clientInfo"; + /** + * Key of the payload of the {@code connection_init} message of a {@code graphql-transport-ws} + * connection, a {@code Map}, absent if the client sent none. + */ + public static final String CONNECTION_INIT_PAYLOAD = + "org.restlet.ext.graphql.connectionInitPayload"; + private RestletGraphql() {} } 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 050fb1698e..dca1712cc5 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 @@ -12,17 +12,13 @@ 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; @@ -53,7 +49,7 @@ public final class GraphqlEventStream implements EventWriter { private final BooleanSupplier endpointStopped; - private volatile QueueingSubscriber subscriber; + private volatile GraphqlResultQueue subscriber; private volatile boolean stopped; @@ -139,9 +135,9 @@ private static ServerSentEvent error(String message) { */ public void stop() { stopped = true; - QueueingSubscriber current = subscriber; + GraphqlResultQueue current = subscriber; if (current != null) { - current.fail(new Failure("The GraphQL endpoint is stopping", null)); + current.fail(new GraphqlResultQueue.Failure("The GraphQL endpoint is stopping", null)); } } @@ -159,7 +155,7 @@ public void write(EventSink sink) throws IOException { sink.send(COMPLETE); return; } - QueueingSubscriber subscriber = new QueueingSubscriber(maxQueuedBytes); + GraphqlResultQueue subscriber = new GraphqlResultQueue(maxQueuedBytes); this.subscriber = subscriber; openStreams.add(this); try { @@ -167,16 +163,17 @@ public void write(EventSink sink) throws IOException { 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)); + subscriber.fail( + new GraphqlResultQueue.Failure("The GraphQL endpoint is stopping", null)); } else { publisher.subscribe(subscriber); } while (true) { - Object signal = subscriber.take(); - if (signal instanceof Next next) { + GraphqlResultQueue.Signal signal = subscriber.take(); + if (signal instanceof GraphqlResultQueue.Next next) { sink.send(next(next.json())); subscriber.request(); - } else if (signal instanceof Failure failure) { + } else if (signal instanceof GraphqlResultQueue.Failure failure) { if (failure.cause() != null) { logger.log(Level.WARNING, "GraphQL subscription failed", failure.cause()); } @@ -198,111 +195,4 @@ public void write(EventSink sink) throws IOException { 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/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java new file mode 100644 index 0000000000..5ff1d76746 --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java @@ -0,0 +1,165 @@ +/** + * 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 java.nio.charset.StandardCharsets; +import java.util.concurrent.LinkedBlockingQueue; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + +/** + * Subscribes to the results of a subscription and queues them, serialized, for the thread that + * writes them. One result is requested at a time: the writer calls {@link #request()} once a result + * is written. Results received without being requested are queued up to a byte limit, beyond which + * the subscription is cancelled. + */ +public final class GraphqlResultQueue implements Subscriber { + + /** A signal taken from the queue. */ + public sealed interface Signal permits Next, Failure, Completed, Cancelled {} + + /** + * A serialized result. + * + * @param json The result, in JSON. + * @param size The size of the JSON text in UTF-8 bytes. + */ + public record Next(String json, long size) implements Signal {} + + /** + * A terminal failure. + * + * @param message The message sent to the client. + * @param cause The cause to log, or null. + */ + public record Failure(String message, Throwable cause) implements Signal {} + + /** The completion of the subscription. */ + public record Completed() implements Signal {} + + /** The cancellation of the subscription by the client: nothing more is sent. */ + public record Cancelled() implements Signal {} + + private final LinkedBlockingQueue signals = new LinkedBlockingQueue<>(); + + private final long maxQueuedBytes; + + private long queuedBytes; + + private volatile Subscription subscription; + + private volatile boolean cancelled; + + private boolean terminated; + + /** + * Constructor. + * + * @param maxQueuedBytes The maximum number of bytes of results received but not yet taken. + */ + public GraphqlResultQueue(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(new Completed()); + } + + /** + * Ends the queue with a failure and cancels the subscription. + * + * @param failure The failure. + */ + public void fail(Failure failure) { + terminate(failure); + cancel(); + } + + /** Ends the queue silently and cancels the subscription, as requested by the client. */ + public void cancelByClient() { + terminate(new Cancelled()); + cancel(); + } + + private synchronized void terminate(Signal signal) { + if (!terminated) { + terminated = true; + signals.add(signal); + } + } + + /** + * Takes the next signal, blocking until one is available. + * + * @return The next signal. + * @throws InterruptedException If interrupted while waiting. + */ + public Signal take() throws InterruptedException { + Signal signal = signals.take(); + if (signal instanceof Next next) { + synchronized (this) { + queuedBytes -= next.size(); + } + } + return signal; + } + + /** Requests the next result, once the previous one is written. */ + public void request() { + Subscription current = subscription; + if (current != null && !cancelled) { + current.request(1); + } + } + + /** Cancels the subscription. Does nothing if already cancelled. */ + public 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/GraphqlModulePathTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlModulePathTestCase.java index 1fc59a162b..460cf1b8f9 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlModulePathTestCase.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlModulePathTestCase.java @@ -10,6 +10,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.ByteArrayOutputStream; import java.io.File; @@ -138,6 +139,7 @@ void namedConsumerResolvesAndRuns(@TempDir Path dir) throws Exception { Set.of("consumer")); assertNotNull(configuration.findModule("com.graphqljava").orElse(null)); assertNotNull(configuration.findModule("org.reactivestreams").orElse(null)); + assertTrue(configuration.findModule("org.restlet.ext.websocket").isEmpty()); ModuleLayer layer = ModuleLayer.boot() @@ -159,7 +161,9 @@ private static List modulePath() { || entry.contains("test-classes") || entry.contains("junit") || entry.contains("opentest4j") - || entry.contains("apiguardian")) { + || entry.contains("apiguardian") + // The WebSocket extension is optional: the consumer must run without it. + || entry.contains("ext.websocket")) { continue; } try { 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 9b74cae46d..e1419ace92 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 @@ -37,6 +37,7 @@ final class GraphqlTestSchemas { loaded(key: String!): String echo(input: EchoInput!): String header(name: String!): String + initToken: String } input EchoInput { text: String! @@ -87,6 +88,12 @@ static GraphQLSchema schema(AtomicInteger counter) { Request request = env.getGraphQlContext().get(RestletGraphql.REQUEST); return request.getHeaders().getFirstValue(env.getArgument("name"), true); }; + DataFetcher initToken = + env -> { + Map payload = + env.getGraphQlContext().get(RestletGraphql.CONNECTION_INIT_PAYLOAD); + return payload == null ? null : payload.get("token"); + }; DataFetcher ticks = env -> new TickPublisher(env.getArgument("count")); RuntimeWiring wiring = @@ -108,7 +115,8 @@ static GraphQLSchema schema(AtomicInteger counter) { .dataFetcher("userAgent", userAgent) .dataFetcher("loaded", loaded) .dataFetcher("echo", echo) - .dataFetcher("header", header)) + .dataFetcher("header", header) + .dataFetcher("initToken", initToken)) .type( "Mutation", type -> diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java new file mode 100644 index 0000000000..653d85990d --- /dev/null +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java @@ -0,0 +1,328 @@ +/** + * 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.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.WebSocket; +import java.net.http.WebSocketHandshakeException; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +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.restlet.Client; +import org.restlet.Component; +import org.restlet.data.Protocol; +import org.restlet.ext.graphql.internal.GraphqlJson; + +/** + * The {@code graphql-transport-ws} server of {@link GraphqlServerAdapter}, driven by the JDK + * WebSocket client. + */ +class GraphqlWebSocketServerTestCase { + + private Component component; + + private GraphqlServerAdapter adapter; + + private HttpClient http; + + private String base; + + @BeforeEach + void setUp() throws Exception { + component = new Component(); + component.getServers().add(Protocol.HTTP, 0); + adapter = new GraphqlServerAdapter(); + adapter.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + adapter.getSubscriptionTransports().add(GraphqlTransport.WEBSOCKET); + adapter.setConnectionInitTimeout(Duration.ofMillis(300)); + component.getDefaultHost().attach("/graphql", adapter); + component.start(); + base = "localhost:" + component.getServers().get(0).getActualPort() + "/graphql"; + http = HttpClient.newHttpClient(); + } + + @AfterEach + void tearDown() throws Exception { + http.close(); + component.stop(); + } + + private Peer connect() throws Exception { + Peer peer = new Peer(); + peer.socket = + http.newWebSocketBuilder() + .subprotocols(GraphqlWebSocketSupport.SUBPROTOCOL) + .buildAsync(URI.create("ws://" + base), peer) + .get(5, TimeUnit.SECONDS); + return peer; + } + + private Peer initialized(Map payload) throws Exception { + Peer peer = connect(); + peer.send( + payload == null + ? Map.of("type", "connection_init") + : Map.of("type", "connection_init", "payload", payload)); + assertEquals(Map.of("type", "connection_ack"), peer.next()); + return peer; + } + + private static Map subscribe(String id, String query) { + return Map.of("id", id, "type", "subscribe", "payload", Map.of("query", query)); + } + + 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); + } + } + + @Test + void handshakeAndPing() throws Exception { + Peer peer = initialized(Map.of("token", "t")); + assertEquals("graphql-transport-ws", peer.socket.getSubprotocol()); + peer.send(Map.of("type", "ping")); + assertEquals(Map.of("type", "pong"), peer.next()); + peer.send(Map.of("type", "pong")); + peer.send(subscribe("1", "{ initToken }")); + assertEquals( + Map.of( + "id", + "1", + "type", + "next", + "payload", + Map.of("data", Map.of("initToken", "t"))), + peer.next()); + assertEquals(Map.of("id", "1", "type", "complete"), peer.next()); + peer.close(); + } + + @Test + void subscriptionStreamsNextThenComplete() throws Exception { + Peer peer = initialized(null); + peer.send(subscribe("s", "subscription { ticks(count: 3) }")); + for (int i = 1; i <= 3; i++) { + assertEquals( + Map.of( + "id", + "s", + "type", + "next", + "payload", + Map.of("data", Map.of("ticks", i))), + peer.next()); + } + assertEquals(Map.of("id", "s", "type", "complete"), peer.next()); + await(() -> adapter.getActiveSubscriptions() == 0); + peer.close(); + } + + @Test + void errorsBeforeExecutionAreErrorMessages() throws Exception { + Peer peer = initialized(null); + peer.send(subscribe("e", "subscription { unknown }")); + Map error = peer.next(); + assertEquals("error", error.get("type")); + assertEquals("e", error.get("id")); + assertInstanceOf(List.class, error.get("payload")); + // No complete after an error: the next message answers a ping. + peer.send(Map.of("type", "ping")); + assertEquals(Map.of("type", "pong"), peer.next()); + peer.close(); + } + + @Test + void severalSubscriptionsPerConnection() throws Exception { + int cancelled = TickPublisher.CANCELLED.get(); + Peer peer = initialized(null); + peer.send(subscribe("a", "subscription { idle }")); + peer.send(subscribe("b", "subscription { idle }")); + await(() -> adapter.getActiveSubscriptions() == 2); + peer.send(Map.of("id", "a", "type", "complete")); + await(() -> adapter.getActiveSubscriptions() == 1); + assertEquals(cancelled + 1, TickPublisher.CANCELLED.get()); + + // The id is free again once completed. + peer.send(subscribe("a", "{ hello }")); + assertEquals("next", peer.next().get("type")); + assertEquals("complete", peer.next().get("type")); + + peer.close(); + await(() -> adapter.getActiveSubscriptions() == 0); + assertEquals(cancelled + 2, TickPublisher.CANCELLED.get()); + } + + @Test + void protocolViolationsClose() throws Exception { + Peer early = connect(); + early.send(subscribe("1", "{ hello }")); + assertEquals(4401, early.closeCode()); + + Peer slow = connect(); + assertEquals(4408, slow.closeCode()); + + Peer twice = initialized(null); + twice.send(Map.of("type", "connection_init")); + assertEquals(4429, twice.closeCode()); + + Peer duplicate = initialized(null); + duplicate.send(subscribe("d", "subscription { idle }")); + duplicate.send(subscribe("d", "subscription { idle }")); + assertEquals(4409, duplicate.closeCode()); + + for (String text : + List.of( + "not json", + "{\"type\": \"unknown\"}", + "{\"id\": \"1\", \"type\": \"subscribe\"}", + "{\"type\": \"complete\"}", + "{\"type\": \"connection_init\", \"payload\": 1}", + "{\"id\": \"1\", \"type\": \"subscribe\", \"payload\": {\"query\": \"{ hello }\", \"variables\": []}}")) { + Peer invalid = text.contains("connection_init") ? connect() : initialized(null); + invalid.socket.sendText(text, true).get(5, TimeUnit.SECONDS); + assertEquals(4400, invalid.closeCode(), text); + } + + Peer binary = initialized(null); + binary.socket.sendBinary(java.nio.ByteBuffer.wrap(new byte[] {1}), true).get(); + assertEquals(4400, binary.closeCode()); + await(() -> adapter.getActiveSubscriptions() == 0); + } + + @Test + void subprotocolIsRequired() { + ExecutionException e = + assertThrows( + ExecutionException.class, + () -> + http.newWebSocketBuilder() + .subprotocols("graphql-ws") + .buildAsync(URI.create("ws://" + base), new Peer()) + .get(5, TimeUnit.SECONDS)); + WebSocketHandshakeException handshake = + assertInstanceOf(WebSocketHandshakeException.class, e.getCause()); + assertEquals(400, handshake.getResponse().statusCode()); + } + + @Test + void httpStillWorksOnTheSamePath() throws Exception { + Client client = new Client(Protocol.HTTP); + client.start(); + try { + GraphqlClientAdapter graphql = new GraphqlClientAdapter(); + graphql.setClient(client); + graphql.setEndpoint("http://" + base); + graphql.start(); + assertEquals( + Map.of("hello", "Hello world"), + graphql.execute("{ hello }", null, null).getData()); + } finally { + client.stop(); + } + } + + @Test + void stoppingTheAdapterClosesConnections() throws Exception { + Peer peer = initialized(null); + peer.send(subscribe("s", "subscription { idle }")); + await(() -> adapter.getActiveSubscriptions() == 1); + adapter.stop(); + assertEquals(1001, peer.closeCode()); + await(() -> adapter.getActiveSubscriptions() == 0); + } + + @Test + void configurationIsValidated() { + GraphqlServerAdapter other = new GraphqlServerAdapter(); + assertEquals( + GraphqlServerAdapter.DEFAULT_CONNECTION_INIT_TIMEOUT, + other.getConnectionInitTimeout()); + assertThrows(IllegalArgumentException.class, () -> other.setConnectionInitTimeout(null)); + assertThrows( + IllegalArgumentException.class, + () -> other.setConnectionInitTimeout(Duration.ZERO)); + assertThrows( + IllegalStateException.class, + () -> adapter.setConnectionInitTimeout(Duration.ofSeconds(1))); + } + + /** A JDK WebSocket listener collecting JSON messages and the close code. */ + private static final class Peer implements WebSocket.Listener { + + final BlockingQueue> messages = new LinkedBlockingQueue<>(); + + final CompletableFuture closed = new CompletableFuture<>(); + + private final StringBuilder text = new StringBuilder(); + + WebSocket socket; + + @Override + public CompletionStage onText(WebSocket webSocket, CharSequence data, boolean last) { + text.append(data); + if (last) { + messages.add(GraphqlJson.asObject(GraphqlJson.parse(text.toString()))); + text.setLength(0); + } + webSocket.request(1); + return null; + } + + @Override + public CompletionStage onClose(WebSocket webSocket, int statusCode, String reason) { + closed.complete(statusCode); + return null; + } + + @Override + public void onError(WebSocket webSocket, Throwable error) { + closed.complete(-1); + } + + void send(Map message) throws Exception { + socket.sendText(GraphqlJson.writeString(message), true).get(5, TimeUnit.SECONDS); + } + + Map next() throws InterruptedException { + Map message = messages.poll(5, TimeUnit.SECONDS); + assertTrue(message != null, "Timed out waiting for a message"); + return message; + } + + int closeCode() throws Exception { + return closed.get(5, TimeUnit.SECONDS); + } + + void close() throws Exception { + socket.sendClose(WebSocket.NORMAL_CLOSURE, "").get(5, TimeUnit.SECONDS); + assertNull(messages.poll(50, TimeUnit.MILLISECONDS)); + } + } +} From fbd19ebd7974f9a25de16649aebeb3e8f478ea2f Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 23:32:45 -0400 Subject: [PATCH 2/3] test: give WebSocket clients 2 s to initialize in GraphQL tests A 300 ms connection_init timeout closed the connection before a slow CI runner sent connection_init. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../restlet/ext/graphql/GraphqlWebSocketServerTestCase.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java index 653d85990d..45c623211e 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java @@ -58,7 +58,8 @@ void setUp() throws Exception { adapter = new GraphqlServerAdapter(); adapter.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); adapter.getSubscriptionTransports().add(GraphqlTransport.WEBSOCKET); - adapter.setConnectionInitTimeout(Duration.ofMillis(300)); + // Long enough for slow CI runners to send connection_init, short enough for the 4408 test. + adapter.setConnectionInitTimeout(Duration.ofSeconds(2)); component.getDefaultHost().attach("/graphql", adapter); component.start(); base = "localhost:" + component.getServers().get(0).getActualPort() + "/graphql"; From 63ed5231a209574bc2e25b2dd2077376aaaa250a Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:18:25 -0400 Subject: [PATCH 3/3] fix: harden the GraphQL WebSocket server lifecycle Close connections accepted while stopping and close them concurrently, reserve a subscription slot before executing a subscription, skip operations cancelled before they start, and retire operation IDs atomically with their messages. Validate ping and pong payloads, bound close reasons to 123 bytes and let connection_init win or lose against the timeout atomically. Co-Authored-By: Claude Opus 5.5 --- .../ext/graphql/GraphqlServerAdapter.java | 34 +++- .../ext/graphql/GraphqlWebSocketSupport.java | 166 +++++++++++++++--- .../GraphqlWebSocketServerTestCase.java | 53 ++++++ 3 files changed, 226 insertions(+), 27 deletions(-) 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 bed7f4cba5..7e25a974bc 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 @@ -548,16 +548,42 @@ public void handle(Request request, Response response) { } /** - * Executes an operation. + * Customizes and builds the execution input of an operation. * * @param request The request. * @param input The execution input, before customization. * @param initPayload The payload of a WebSocket connection_init message, or null. - * @return The result. + * @return The execution input. */ - ExecutionResult execute( + ExecutionInput prepare( Request request, ExecutionInput.Builder input, Map initPayload) { - return graphQL.execute(customize(request, input, initPayload).build()); + return customize(request, input, initPayload).build(); + } + + /** + * Indicates if an execution input selects a subscription. Invalid documents are left to the + * execution, which reports their errors. + * + * @param input The execution input. + * @return True if the selected operation is a subscription. + */ + boolean isSubscription(ExecutionInput input) { + try { + return selectOperation(input.getQuery(), input.getOperationName()) + == OperationDefinition.Operation.SUBSCRIPTION; + } catch (InvalidSyntaxException | RequestException _) { + return false; + } + } + + /** + * Executes an operation. + * + * @param input The execution input, customized. + * @return The result. + */ + ExecutionResult execute(ExecutionInput input) { + return graphQL.execute(input); } private ExecutionInput.Builder customize( diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java index c193f8903b..4cbb9545ba 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java @@ -13,6 +13,9 @@ import graphql.GraphQLError; import graphql.GraphqlErrorBuilder; import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -51,10 +54,18 @@ final class GraphqlWebSocketSupport implements GraphqlUpgradeSupport { static final int TOO_MANY_INIT = 4429; + /** The maximum size of a close reason, in UTF-8 bytes. */ + private static final int MAX_CLOSE_REASON_BYTES = 123; + + /** The time given to the open connections to close when stopping. */ + private static final Duration CLOSE_TIMEOUT = Duration.ofSeconds(5); + private final GraphqlServerAdapter adapter; private final Set connections = ConcurrentHashMap.newKeySet(); + private volatile boolean stopped; + GraphqlWebSocketSupport(GraphqlServerAdapter adapter) { this.adapter = adapter; } @@ -77,11 +88,53 @@ public boolean handle(Request request, Response response) { @Override public void stop() { + // Set before the snapshot, so that connections registered later close themselves. + stopped = true; + // Closed concurrently, as each close waits for the peer's close handshake. + List closing = new ArrayList<>(); for (Connection connection : List.copyOf(connections)) { - connection.close(WebSocketSession.GOING_AWAY, "The GraphQL endpoint is stopping"); + closing.add( + Thread.ofVirtual() + .name("restlet-graphql-ws-close") + .start(connection::closeGoingAway)); + } + long deadline = System.nanoTime() + CLOSE_TIMEOUT.toNanos(); + try { + for (Thread thread : closing) { + long remaining = deadline - System.nanoTime(); + if (remaining <= 0 || !thread.join(Duration.ofNanos(remaining))) { + break; + } + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); } } + /** Truncates a close reason to the size allowed in a close frame. */ + static String closeReason(String reason) { + if (reason == null + || reason.getBytes(StandardCharsets.UTF_8).length <= MAX_CLOSE_REASON_BYTES) { + return reason; + } + StringBuilder result = new StringBuilder(); + int bytes = 0; + for (int i = 0; i < reason.length(); ) { + int codePoint = reason.codePointAt(i); + int size = + new String(Character.toChars(codePoint)) + .getBytes(StandardCharsets.UTF_8) + .length; + if (bytes + size > MAX_CLOSE_REASON_BYTES) { + break; + } + result.appendCodePoint(codePoint); + bytes += size; + i += Character.charCount(codePoint); + } + return result.toString(); + } + private static String message(String type, String id, Object payload) { Map message = new LinkedHashMap<>(); if (id != null) { @@ -113,6 +166,10 @@ private final class Connection { private final Object sendLock = new Object(); + private final Object initLock = new Object(); + + private boolean initTimedOut; + private volatile boolean initialized; private volatile Map initPayload; @@ -124,6 +181,12 @@ private final class Connection { void run() throws IOException { connections.add(this); + if (stopped) { + // Accepted while the endpoint was stopping, after its connections were closed. + connections.remove(this); + closeGoingAway(); + return; + } Thread timeout = Thread.ofVirtual().name("restlet-graphql-ws-init").start(this::awaitInit); try { @@ -160,9 +223,13 @@ private void awaitInit() { } catch (InterruptedException _) { return; } - if (!initialized) { - close(INIT_TIMEOUT, "Connection initialisation timeout"); + synchronized (initLock) { + if (initialized) { + return; + } + initTimedOut = true; } + close(INIT_TIMEOUT, "Connection initialisation timeout"); } /** Handles a message. Returns false once the connection is closed. */ @@ -180,14 +247,25 @@ private boolean dispatch(Map message) { if (payload != null && GraphqlJson.asObject(payload) == null) { return close(INVALID_MESSAGE, "Invalid connection_init payload"); } - initPayload = GraphqlJson.asObject(payload); - initialized = true; + synchronized (initLock) { + if (initTimedOut) { + return false; + } + initPayload = GraphqlJson.asObject(payload); + initialized = true; + } return send(message("connection_ack", null, null)); } case "ping" -> { + if (invalidObject(message, "payload")) { + return close(INVALID_MESSAGE, "Invalid ping payload"); + } return send(message("pong", null, null)); } case "pong" -> { + if (invalidObject(message, "payload")) { + return close(INVALID_MESSAGE, "Invalid pong payload"); + } return true; } case "subscribe" -> { @@ -262,10 +340,32 @@ boolean send(String text) { } } + /** + * Sends a message of an operation, unless the operation is no longer registered under its + * ID, so that a reused ID never receives the messages of the operation it replaces. A + * terminal message unregisters the operation first, so the ID can be reused as soon as it + * is received. + * + * @return False if the operation is retired or the session is closed. + */ + boolean send(Operation operation, String text, boolean terminal) { + synchronized (sendLock) { + boolean registered = + terminal + ? operations.remove(operation.id, operation) + : operations.get(operation.id) == operation; + return registered && send(text); + } + } + + void closeGoingAway() { + close(WebSocketSession.GOING_AWAY, "The GraphQL endpoint is stopping"); + } + /** Closes the connection. Returns false, as a convenience for the dispatch loop. */ boolean close(int code, String reason) { try { - session.close(code, reason); + session.close(code, closeReason(reason)); } catch (IOException e) { adapter.getLogger().log(Level.FINE, "Unable to close a GraphQL WebSocket", e); } @@ -302,36 +402,56 @@ void cancel() { @Override public void run() { + AtomicInteger active = adapter.getActiveSubscriptionCounter(); + boolean reserved = false; try { - ExecutionResult result = - adapter.execute(connection.request, input, connection.initPayload); + // A complete message or a closed connection can come before this thread starts. + if (cancelled) { + return; + } + ExecutionInput executionInput = + adapter.prepare(connection.request, input, connection.initPayload); + if (adapter.isSubscription(executionInput)) { + // Reserved before execution, which sets up the subscription. + if (active.incrementAndGet() > adapter.getMaxSubscriptions()) { + active.decrementAndGet(); + send(message("error", id, error("Too many active subscriptions")), true); + return; + } + reserved = true; + } + if (cancelled) { + return; + } + ExecutionResult result = adapter.execute(executionInput); if (cancelled) { return; } if (!result.isDataPresent()) { - connection.send(message("error", id, errors(result.getErrors()))); + send(message("error", id, errors(result.getErrors())), true); } else if (result.getData() instanceof Publisher publisher) { @SuppressWarnings("unchecked") Publisher results = (Publisher) publisher; stream(results); - } else if (connection.send(message("next", id, result.toSpecification()))) { - connection.send(message("complete", id, null)); + } else if (send(message("next", id, result.toSpecification()), false)) { + send(message("complete", id, null), true); } } catch (RuntimeException e) { adapter.getLogger().log(Level.WARNING, "GraphQL operation failed", e); - connection.send(message("error", id, error("The operation failed"))); + send(message("error", id, error("The operation failed")), true); } finally { + if (reserved) { + active.decrementAndGet(); + } connection.operations.remove(id, this); } } + private boolean send(String text, boolean terminal) { + return connection.send(this, text, terminal); + } + private void stream(Publisher results) { - AtomicInteger active = adapter.getActiveSubscriptionCounter(); - if (active.incrementAndGet() > adapter.getMaxSubscriptions()) { - active.decrementAndGet(); - connection.send(message("error", id, error("Too many active subscriptions"))); - return; - } GraphqlResultQueue current = new GraphqlResultQueue(adapter.getMaxQueuedBytes()); queue = current; try { @@ -342,12 +462,13 @@ private void stream(Publisher results) { while (true) { GraphqlResultQueue.Signal signal = current.take(); if (signal instanceof GraphqlResultQueue.Next next) { - if (!connection.send( + if (!send( "{\"id\":" + GraphqlJson.writeString(id) + ",\"type\":\"next\",\"payload\":" + next.json() - + "}")) { + + "}", + false)) { return; } current.request(); @@ -359,10 +480,10 @@ private void stream(Publisher results) { "GraphQL subscription failed", failure.cause()); } - connection.send(message("error", id, error(failure.message()))); + send(message("error", id, error(failure.message())), true); return; } else if (signal instanceof GraphqlResultQueue.Completed) { - connection.send(message("complete", id, null)); + send(message("complete", id, null), true); return; } else { return; @@ -372,7 +493,6 @@ private void stream(Publisher results) { Thread.currentThread().interrupt(); } finally { current.cancel(); - active.decrementAndGet(); } } } diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java index 45c623211e..5bbe5cf611 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java @@ -259,6 +259,59 @@ void stoppingTheAdapterClosesConnections() throws Exception { await(() -> adapter.getActiveSubscriptions() == 0); } + @Test + void invalidPayloadsAndLongReasonsClose() throws Exception { + for (String text : + List.of( + "{\"type\": \"ping\", \"payload\": []}", + "{\"type\": \"pong\", \"payload\": \"x\"}", + "{\"type\": \"" + "x".repeat(300) + "\"}")) { + Peer invalid = initialized(null); + invalid.socket.sendText(text, true).get(5, TimeUnit.SECONDS); + assertEquals(4400, invalid.closeCode(), text); + } + // Truncated to whole characters within 123 UTF-8 bytes. + assertEquals("é".repeat(61), GraphqlWebSocketSupport.closeReason("é".repeat(100))); + assertEquals("short", GraphqlWebSocketSupport.closeReason("short")); + } + + @Test + void subscriptionLimitIsCheckedBeforeExecution() throws Exception { + GraphqlServerAdapter limited = new GraphqlServerAdapter(); + limited.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + limited.getSubscriptionTransports().add(GraphqlTransport.WEBSOCKET); + limited.setMaxSubscriptions(1); + component.getDefaultHost().attach("/limited", limited); + limited.start(); + Peer peer = new Peer(); + peer.socket = + http.newWebSocketBuilder() + .subprotocols(GraphqlWebSocketSupport.SUBPROTOCOL) + .buildAsync( + URI.create("ws://" + base.replace("/graphql", "/limited")), peer) + .get(5, TimeUnit.SECONDS); + peer.send(Map.of("type", "connection_init")); + assertEquals(Map.of("type", "connection_ack"), peer.next()); + + // An ID completed by the server can be reused at once. + for (int i = 0; i < 20; i++) { + peer.send(subscribe("q", "{ hello }")); + assertEquals("next", peer.next().get("type")); + assertEquals(Map.of("id", "q", "type", "complete"), peer.next()); + } + + peer.send(subscribe("a", "subscription { idle }")); + await(() -> limited.getActiveSubscriptions() == 1); + int created = TickPublisher.CREATED.get(); + peer.send(subscribe("b", "subscription { idle }")); + Map error = peer.next(); + assertEquals("error", error.get("type")); + assertEquals("b", error.get("id")); + assertEquals(created, TickPublisher.CREATED.get()); + peer.close(); + limited.stop(); + } + @Test void configurationIsValidated() { GraphqlServerAdapter other = new GraphqlServerAdapter();