From cbc393e339ce75cd67139825eee2ff1bd86a7ea5 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 23:05:25 -0400 Subject: [PATCH 1/3] feat: add the GraphQL WebSocket client (G4, graphql-transport-ws) - GraphqlClientAdapter.setSubscriptionTransport(WEBSOCKET): subscribe() multiplexes operations over one graphql-transport-ws connection, opened on the first demand at the endpoint with the ws or wss scheme. - The WebSocketClient borrows the event loop of the adapter's HTTP client (WebSocketClient.setClient, #1567), so no extra Netty threads are created. The request customizer adds handshake headers and setConnectionInitPayload sets the connection_init payload. - Each operation is delivered by its own virtual thread, honoring demand. The protocol has no flow control: results received without demand are buffered up to maxResponseBytes per operation, beyond which the operation fails and the server is sent complete. - Errors before execution are a single result; a closed connection fails its operations with the close code; cancel() sends complete; stop() fails open operations, closes the connection and stops the WebSocket client, but not the borrowed HTTP client. - The WebSocket code is only loaded when the transport is selected. Refs #1586 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 4 +- .../ext/graphql/GraphqlClientAdapter.java | 75 ++- .../graphql/GraphqlSubscriptionTransport.java | 36 ++ .../graphql/GraphqlWebSocketTransport.java | 476 ++++++++++++++++++ .../GraphqlWebSocketClientTestCase.java | 270 ++++++++++ 5 files changed, 857 insertions(+), 4 deletions(-) create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSubscriptionTransport.java create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.java create mode 100644 org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java diff --git a/changes.md b/changes.md index b11842b2dd..7b6138f4e4 100644 --- a/changes.md +++ b/changes.md @@ -162,7 +162,9 @@ Changes log the server adapter also accepts `graphql-transport-ws` WebSocket upgrades on its own path (HTTP/1.1 only), with several operations per connection, a `connection_init` timeout and the protocol's close codes; `org.restlet.ext.websocket` is an optional dependency, - loaded only then. Incremental delivery of `@defer` results over `multipart/mixed` is + loaded only then. `GraphqlClientAdapter.setSubscriptionTransport(WEBSOCKET)` multiplexes + subscriptions over one `graphql-transport-ws` connection sharing the borrowed client's + event loop. Incremental delivery of `@defer` results over `multipart/mixed` is available on both sides when enabled (experimental, off by default): the client returns graphql-java's `IncrementalExecutionResult`, whose item publisher reads the delayed results. Issue #1581. 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 a893286f07..870d8e8ecf 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 @@ -104,6 +104,12 @@ private enum State { private volatile boolean incrementalDeliveryEnabled; + private volatile GraphqlTransport subscriptionTransport = GraphqlTransport.SSE; + + private volatile Map connectionInitPayload; + + private volatile GraphqlSubscriptionTransport webSocketTransport; + private final Set subscriptions = ConcurrentHashMap.newKeySet(); @@ -282,6 +288,50 @@ public void setIncrementalDeliveryEnabled(boolean incrementalDeliveryEnabled) { this.incrementalDeliveryEnabled = incrementalDeliveryEnabled; } + /** + * Returns the transport of subscriptions. + * + * @return The transport of subscriptions. + */ + public GraphqlTransport getSubscriptionTransport() { + return subscriptionTransport; + } + + /** + * Sets the transport of subscriptions. Defaults to {@link GraphqlTransport#SSE}, with one + * request per subscription. With {@link GraphqlTransport#WEBSOCKET}, subscriptions are + * multiplexed over one {@code graphql-transport-ws} connection, opened on the first demand at + * the endpoint URI with the "ws" or "wss" scheme. It shares the event loop of the borrowed + * client, which must then be a started Netty HTTP client, and requires the {@code + * org.restlet.ext.websocket} module. The request customizer adds handshake headers. + * + * @param subscriptionTransport The transport of subscriptions. + */ + public void setSubscriptionTransport(GraphqlTransport subscriptionTransport) { + checkNew(); + this.subscriptionTransport = Objects.requireNonNull(subscriptionTransport); + } + + /** + * Returns the payload of the {@code connection_init} message of WebSocket connections. + * + * @return The payload, or null. + */ + public Map getConnectionInitPayload() { + return connectionInitPayload; + } + + /** + * Sets the payload of the {@code connection_init} message of WebSocket connections, for example + * an authentication token. + * + * @param connectionInitPayload The payload, or null. + */ + public void setConnectionInitPayload(Map connectionInitPayload) { + checkNew(); + this.connectionInitPayload = connectionInitPayload; + } + /** * Validates and freezes the configuration. * @@ -299,6 +349,14 @@ public synchronized void start() { if (parserOptions == null) { parserOptions = ParserOptions.getDefaultOperationParserOptions(); } + if (subscriptionTransport == GraphqlTransport.WEBSOCKET) { + try { + // Only loads the optional WebSocket extension when selected. + webSocketTransport = new GraphqlWebSocketTransport(this); + } catch (IOException e) { + throw new IllegalStateException("Unable to start the WebSocket client", e); + } + } state = State.STARTED; } @@ -312,6 +370,10 @@ public synchronized void stop() { List.copyOf(subscriptions)) { subscription.stop(); } + GraphqlSubscriptionTransport transport = webSocketTransport; + if (transport != null) { + transport.stop(); + } } /** @@ -388,8 +450,7 @@ public Publisher subscribe( ExecutionResult invalid = validate(document); return invalid != null ? single(invalid) - : new GraphqlSsePublisher( - this, AstPrinter.printAst(document), operationName, variables); + : publisher(AstPrinter.printAst(document), operationName, variables); } /** @@ -423,7 +484,15 @@ public Publisher subscribe( return single(invalid); } } - return new GraphqlSsePublisher(this, document, operationName, variables); + return publisher(document, operationName, variables); + } + + private Publisher publisher( + String document, String operationName, Map variables) { + GraphqlSubscriptionTransport transport = webSocketTransport; + return transport != null + ? transport.subscribe(document, operationName, variables) + : new GraphqlSsePublisher(this, document, operationName, variables); } /** A publisher of one result, for documents that fail local validation. */ diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSubscriptionTransport.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSubscriptionTransport.java new file mode 100644 index 0000000000..f705d7ffdc --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSubscriptionTransport.java @@ -0,0 +1,36 @@ +/** + * 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.util.Map; +import org.reactivestreams.Publisher; + +/** + * A subscription transport of {@link GraphqlClientAdapter} other than SSE. The implementation + * depends on the optional WebSocket extension: the adapter only refers to it through this + * interface, so that the extension is loaded only when {@link GraphqlTransport#WEBSOCKET} is + * selected. + */ +interface GraphqlSubscriptionTransport { + + /** + * Subscribes to an operation. + * + * @param document The document. + * @param operationName The name of the operation to execute, or null. + * @param variables The variables, or null. + * @return A cold publisher of the results. + */ + Publisher subscribe( + String document, String operationName, Map variables); + + /** Fails the open subscriptions and closes the connection. */ + void stop(); +} diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.java new file mode 100644 index 0000000000..848ca25d74 --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.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 graphql.ExecutionResult; +import graphql.GraphQLError; +import java.io.IOException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.Request; +import org.restlet.data.Header; +import org.restlet.data.Method; +import org.restlet.ext.graphql.internal.GraphqlJson; +import org.restlet.ext.websocket.WebSocketClient; +import org.restlet.ext.websocket.WebSocketMessage; +import org.restlet.ext.websocket.WebSocketSession; + +/** + * Sends the subscriptions of a {@link GraphqlClientAdapter} with {@code graphql-transport-ws}, all + * multiplexed over one WebSocket connection opened on the first demand. The {@link WebSocketClient} + * shares the event loop of the adapter's borrowed HTTP client. + * + *

The protocol has no flow control: results received without demand are buffered, up to the + * adapter's maximum response size per subscription, beyond which the subscription fails. + */ +final class GraphqlWebSocketTransport implements GraphqlSubscriptionTransport { + + private static final Duration ACK_TIMEOUT = Duration.ofSeconds(10); + + private final GraphqlClientAdapter adapter; + + private final WebSocketClient webSocketClient; + + private final AtomicLong ids = new AtomicLong(); + + private final Map operations = new ConcurrentHashMap<>(); + + private Connection connection; + + private boolean stopped; + + GraphqlWebSocketTransport(GraphqlClientAdapter adapter) throws IOException { + this.adapter = adapter; + this.webSocketClient = new WebSocketClient(); + webSocketClient.setClient(adapter.getClient()); + webSocketClient.setMaxMessageSize( + (int) Math.min(adapter.getMaxResponseBytes(), Integer.MAX_VALUE)); + webSocketClient.start(); + } + + @Override + public Publisher subscribe( + String document, String operationName, Map variables) { + Map payload = new LinkedHashMap<>(); + payload.put("query", document); + if (operationName != null) { + payload.put("operationName", operationName); + } + if (variables != null && !variables.isEmpty()) { + payload.put("variables", variables); + } + return subscriber -> { + Objects.requireNonNull(subscriber, "subscriber"); + subscriber.onSubscribe(new Operation(subscriber, payload)); + }; + } + + @Override + public void stop() { + Connection current; + synchronized (this) { + stopped = true; + current = connection; + connection = null; + } + GraphqlTransportException error = + new GraphqlTransportException("The GraphQL client was stopped", null, null, null); + for (Operation operation : List.copyOf(operations.values())) { + operation.fail(error); + } + if (current != null) { + current.close(WebSocketSession.NORMAL_CLOSURE, null); + } + webSocketClient.stop(); + } + + /** Returns the open connection, opening one if needed. */ + private synchronized Connection connection() throws GraphqlTransportException { + if (stopped) { + throw new GraphqlTransportException("The GraphQL client was stopped", null, null, null); + } + if (connection == null || !connection.session.isOpen()) { + connection = open(); + } + return connection; + } + + private Connection open() throws GraphqlTransportException { + URI uri = URI.create(adapter.getEndpoint().replaceFirst("^http", "ws")); + // The request customizer adds handshake headers, such as an API key. + Request handshake = new Request(Method.GET, adapter.getEndpoint()); + if (adapter.getRequestCustomizer() != null) { + adapter.getRequestCustomizer().accept(handshake); + } + Map headers = new LinkedHashMap<>(); + for (Header header : handshake.getHeaders()) { + headers.put(header.getName(), header.getValue()); + } + + WebSocketSession session; + try { + session = + webSocketClient.connect( + uri, List.of(GraphqlWebSocketSupport.SUBPROTOCOL), headers); + } catch (IOException e) { + throw new GraphqlTransportException( + "Unable to open the GraphQL WebSocket", null, null, e); + } + Connection opened = new Connection(session); + if (!GraphqlWebSocketSupport.SUBPROTOCOL.equals(session.getSubprotocol())) { + opened.close(WebSocketSession.PROTOCOL_ERROR, null); + throw new GraphqlTransportException( + "The server did not accept the graphql-transport-ws subprotocol", + null, + null, + null); + } + Thread.ofVirtual().name("restlet-graphql-ws-client").start(opened::read); + Map init = new LinkedHashMap<>(); + init.put("type", "connection_init"); + if (adapter.getConnectionInitPayload() != null) { + init.put("payload", adapter.getConnectionInitPayload()); + } + try { + opened.send(GraphqlJson.writeString(init)); + opened.acknowledged.get(ACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + } catch (IOException | ExecutionException | TimeoutException e) { + opened.close(WebSocketSession.NORMAL_CLOSURE, null); + throw new GraphqlTransportException( + "The GraphQL WebSocket was not acknowledged", null, null, e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + opened.close(WebSocketSession.NORMAL_CLOSURE, null); + throw new GraphqlTransportException( + "Interrupted while opening the GraphQL WebSocket", null, null, e); + } + return opened; + } + + /** One WebSocket connection, read by a virtual thread. */ + private final class Connection { + + private final WebSocketSession session; + + private final CompletableFuture acknowledged = new CompletableFuture<>(); + + private final Object sendLock = new Object(); + + Connection(WebSocketSession session) { + this.session = session; + } + + void send(String text) throws IOException { + synchronized (sendLock) { + session.sendText(text); + } + } + + void close(int code, String reason) { + try { + session.close(code, reason); + } catch (IOException _) { + // Closing anyway + } + } + + void read() { + try { + WebSocketMessage received; + while ((received = session.receive()) != null) { + if (received.isText()) { + dispatch(received.getText()); + } + } + } catch (IOException | RuntimeException _) { + // Reported below as a closed connection + } + GraphqlTransportException closed = + new GraphqlTransportException( + "The GraphQL WebSocket closed: " + + session.getCloseCode() + + (session.getCloseReason() == null + ? "" + : " " + session.getCloseReason()), + null, + null, + null); + acknowledged.completeExceptionally(closed); + for (Operation operation : List.copyOf(operations.values())) { + if (operation.connection == this) { + operation.fail(closed); + } + } + } + + private void dispatch(String text) throws IOException { + Map message = GraphqlJson.asObject(GraphqlJson.parse(text)); + Object type = message == null ? null : message.get("type"); + Operation operation = + message != null && message.get("id") instanceof String id + ? operations.get(id) + : null; + if ("connection_ack".equals(type)) { + acknowledged.complete(null); + } else if ("ping".equals(type)) { + send("{\"type\":\"pong\"}"); + } else if (operation != null && "next".equals(type)) { + Map payload = GraphqlJson.asObject(message.get("payload")); + operation.next( + ExecutionResult.fromSpecification(payload == null ? Map.of() : payload), + text.getBytes(StandardCharsets.UTF_8).length); + } else if (operation != null && "error".equals(type)) { + List errors = + message.get("payload") instanceof List list + ? list.stream() + .map(GraphqlJson::asObject) + .filter(Objects::nonNull) + .map(GraphQLError::fromSpecification) + .toList() + : List.of(); + operation.error(ExecutionResult.newExecutionResult().errors(errors).build()); + } else if (operation != null && "complete".equals(type)) { + operation.complete(); + } + } + } + + /** A signal for the delivery thread of an operation. */ + private record Signal(ExecutionResult result, long size, boolean last, Throwable error) {} + + /** One operation, delivered to its subscriber by its own virtual thread. */ + private final class Operation implements Subscription { + + private final Subscriber subscriber; + + private final Map payload; + + private final LinkedBlockingQueue inbox = new LinkedBlockingQueue<>(); + + private final String id = Long.toString(ids.incrementAndGet()); + + private volatile Connection connection; + + private long demand; + + private long bufferedBytes; + + private boolean started; + + private boolean invalidDemand; + + private volatile boolean cancelled; + + private volatile boolean terminated; + + private Throwable failure; + + Operation(Subscriber subscriber, Map payload) { + this.subscriber = subscriber; + this.payload = payload; + } + + @Override + public void request(long n) { + synchronized (this) { + if (cancelled) { + return; + } + if (n <= 0) { + invalidDemand = true; + } else { + demand = demand + n < 0 ? Long.MAX_VALUE : demand + n; + } + notifyAll(); + if (started) { + return; + } + started = true; + } + Thread.ofVirtual().name("restlet-graphql-ws-operation").start(this::run); + } + + @Override + public void cancel() { + synchronized (this) { + if (cancelled) { + return; + } + cancelled = true; + notifyAll(); + } + inbox.add(new Signal(null, 0, true, null)); + if (operations.remove(id, this) && !terminated) { + Connection current = connection; + if (current != null) { + try { + current.send("{\"id\":\"" + id + "\",\"type\":\"complete\"}"); + } catch (IOException _) { + // The connection is closing + } + } + } + } + + void next(ExecutionResult result, long size) throws IOException { + synchronized (this) { + bufferedBytes += size; + if (bufferedBytes <= adapter.getMaxResponseBytes()) { + inbox.add(new Signal(result, size, false, null)); + return; + } + } + // Still registered: tell the server to complete the subscription. + boolean registered = operations.remove(id, this); + fail( + new GraphqlTransportException( + "The subscription exceeded " + + adapter.getMaxResponseBytes() + + " buffered bytes", + null, + null, + null)); + if (registered) { + connection.send("{\"id\":\"" + id + "\",\"type\":\"complete\"}"); + } + } + + void error(ExecutionResult result) { + terminated = true; + operations.remove(id, this); + inbox.add(new Signal(result, 0, true, null)); + } + + void complete() { + terminated = true; + operations.remove(id, this); + inbox.add(new Signal(null, 0, true, null)); + } + + /** Fails the operation at once, even ahead of buffered results. */ + void fail(Throwable error) { + terminated = true; + operations.remove(id, this); + synchronized (this) { + if (failure == null) { + failure = error; + } + notifyAll(); + } + inbox.add(new Signal(null, 0, true, error)); + } + + private void run() { + try { + Connection current = connection(); + connection = current; + operations.put(id, this); + if (cancelled) { + operations.remove(id, this); + return; + } + Map subscribe = new LinkedHashMap<>(); + subscribe.put("id", id); + subscribe.put("type", "subscribe"); + subscribe.put("payload", payload); + current.send(GraphqlJson.writeString(subscribe)); + deliver(); + } catch (GraphqlTransportException e) { + operations.remove(id, this); + signalError(e); + } catch (IOException e) { + operations.remove(id, this); + signalError( + new GraphqlTransportException( + "Unable to send the GraphQL subscription", null, null, e)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private void deliver() throws InterruptedException { + while (true) { + Signal signal = inbox.take(); + synchronized (this) { + bufferedBytes -= signal.size(); + } + if (cancelled) { + return; + } + if (signal.error() != null) { + signalError(signal.error()); + return; + } + if (signal.result() != null) { + if (!awaitDemand()) { + Throwable error = failure(); + if (error != null) { + signalError(error); + } + return; + } + subscriber.onNext(signal.result()); + } + if (signal.last()) { + subscriber.onComplete(); + return; + } + } + } + + private void signalError(Throwable error) { + if (!cancelled) { + cancelled = true; + subscriber.onError(error); + } + } + + private synchronized Throwable failure() { + return failure; + } + + /** Waits for demand. Returns false if cancelled or failed meanwhile. */ + private boolean awaitDemand() throws InterruptedException { + synchronized (this) { + while (!cancelled && failure == null && demand == 0 && !invalidDemand) { + wait(); + } + if (cancelled || failure != null) { + return false; + } + if (invalidDemand) { + cancelled = true; + subscriber.onError( + new IllegalArgumentException( + "The requested number of results must be positive")); + return false; + } + if (demand != Long.MAX_VALUE) { + demand--; + } + return true; + } + } + } +} diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java new file mode 100644 index 0000000000..c7a1a308dd --- /dev/null +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java @@ -0,0 +1,270 @@ +/** + * 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 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.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.Client; +import org.restlet.Component; +import org.restlet.Context; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.Protocol; +import org.restlet.ext.websocket.WebSocketUpgrade; +import org.restlet.routing.Filter; + +/** {@link GraphqlClientAdapter} subscriptions over {@code graphql-transport-ws}. */ +class GraphqlWebSocketClientTestCase { + + private final AtomicInteger upgrades = new AtomicInteger(); + + private Component component; + + private GraphqlServerAdapter server; + + private Client client; + + private String base; + + @BeforeEach + void setUp() throws Exception { + component = new Component(); + component.getServers().add(Protocol.HTTP, 0); + server = new GraphqlServerAdapter(); + server.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + server.getSubscriptionTransports().add(GraphqlTransport.WEBSOCKET); + Filter counter = + new Filter() { + @Override + protected int beforeHandle(Request request, Response response) { + if (WebSocketUpgrade.isRequested(request)) { + upgrades.incrementAndGet(); + } + return CONTINUE; + } + }; + counter.setNext(server); + component.getDefaultHost().attach("/graphql", counter); + + GraphqlServerAdapter httpOnly = new GraphqlServerAdapter(); + httpOnly.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + component.getDefaultHost().attach("/http", httpOnly); + component.start(); + base = "http://localhost:" + component.getServers().get(0).getActualPort(); + + // The WebSocket client borrows the event loop of a client having a context. + client = new Client(new Context(), Protocol.HTTP); + client.start(); + } + + @AfterEach + void tearDown() throws Exception { + client.stop(); + component.stop(); + } + + private GraphqlClientAdapter graphql(String path) { + GraphqlClientAdapter graphql = new GraphqlClientAdapter(); + graphql.setClient(client); + graphql.setEndpoint(base + path); + graphql.setSubscriptionTransport(GraphqlTransport.WEBSOCKET); + 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); + } + } + + @Test + void subscriptionsShareOneConnectionAndHonorDemand() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.setConnectionInitPayload(Map.of("token", "secret")); + graphql.setRequestCustomizer(request -> request.getHeaders().add("X-Api-Key", "k1")); + graphql.start(); + + TestSubscriber ticks = new TestSubscriber(); + graphql.subscribe("subscription { ticks(count: 3) }", null, null).subscribe(ticks); + ticks.subscription.request(1); + assertEquals(Map.of("ticks", 1), ticks.next().getData()); + Thread.sleep(100); + assertTrue(ticks.items.isEmpty(), "No result without demand"); + ticks.subscription.request(2); + assertEquals(Map.of("ticks", 2), ticks.next().getData()); + assertEquals(Map.of("ticks", 3), ticks.next().getData()); + assertTrue(ticks.done.await(5, TimeUnit.SECONDS)); + assertNull(ticks.error.get()); + + TestSubscriber query = new TestSubscriber(); + graphql.subscribe("{ initToken header(name: \"X-Api-Key\") }", null, null).subscribe(query); + query.subscription.request(Long.MAX_VALUE); + assertEquals(Map.of("initToken", "secret", "header", "k1"), query.next().getData()); + assertTrue(query.done.await(5, TimeUnit.SECONDS)); + assertEquals(1, upgrades.get()); + graphql.stop(); + } + + @Test + void cancellationCompletesTheServerSubscription() throws Exception { + int cancelled = TickPublisher.CANCELLED.get(); + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber first = new TestSubscriber(); + TestSubscriber second = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(first); + graphql.subscribe("subscription { idle }", null, null).subscribe(second); + first.subscription.request(1); + second.subscription.request(1); + await(() -> server.getActiveSubscriptions() == 2); + + first.subscription.cancel(); + await(() -> server.getActiveSubscriptions() == 1); + assertEquals(cancelled + 1, TickPublisher.CANCELLED.get()); + assertEquals(1, first.done.getCount()); + + graphql.stop(); + assertTrue(second.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, second.error.get()); + await(() -> server.getActiveSubscriptions() == 0); + assertTrue(client.isStarted(), "The borrowed client must survive stop()"); + assertThrows( + IllegalStateException.class, + () -> graphql.subscribe("subscription { idle }", null, null)); + } + + @Test + void errorsBeforeExecutionAreResults() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { unknown }", null, null).subscribe(subscriber); + subscriber.subscription.request(1); + ExecutionResult result = subscriber.next(); + assertFalse(result.isDataPresent()); + assertFalse(result.getErrors().isEmpty()); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertNull(subscriber.error.get()); + graphql.stop(); + } + + @Test + void bufferedResultsAreBounded() throws Exception { + int cancelled = TickPublisher.CANCELLED.get(); + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.setMaxResponseBytes(512); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { ticks(count: -1) }", null, null).subscribe(subscriber); + subscriber.subscription.request(1); + subscriber.next(); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, subscriber.error.get()); + await(() -> TickPublisher.CANCELLED.get() > cancelled); + graphql.stop(); + } + + @Test + void serverClosureFailsSubscriptions() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(subscriber); + subscriber.subscription.request(1); + await(() -> server.getActiveSubscriptions() == 1); + server.stop(); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + GraphqlTransportException e = + assertInstanceOf(GraphqlTransportException.class, subscriber.error.get()); + assertTrue(e.getMessage().contains("1001"), e.getMessage()); + graphql.stop(); + } + + @Test + void endpointsWithoutWebSocketFail() throws Exception { + GraphqlClientAdapter graphql = graphql("/http"); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(subscriber); + subscriber.subscription.request(1); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, subscriber.error.get()); + graphql.stop(); + } + + @Test + void webSocketNeedsAStartedNettyClient() { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.setClient(new Client(Protocol.HTTP)); + assertThrows(IllegalStateException.class, graphql::start); + assertEquals(GraphqlTransport.WEBSOCKET, graphql.getSubscriptionTransport()); + assertNull(graphql.getConnectionInitPayload()); + assertThrows(NullPointerException.class, () -> graphql.setSubscriptionTransport(null)); + } + + /** 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; + } + } +} From 07b94d551496501e73de523889b219f2280d8734 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 23:38:22 -0400 Subject: [PATCH 2/3] test: don't require a result before a GraphQL buffer overflow The client signals the overflow at once, possibly before delivering the first buffered result. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java index c7a1a308dd..3378795e13 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java @@ -184,7 +184,7 @@ void bufferedResultsAreBounded() throws Exception { TestSubscriber subscriber = new TestSubscriber(); graphql.subscribe("subscription { ticks(count: -1) }", null, null).subscribe(subscriber); subscriber.subscription.request(1); - subscriber.next(); + // The failure is signalled at once, possibly before the first buffered result. assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); assertInstanceOf(GraphqlTransportException.class, subscriber.error.get()); await(() -> TickPublisher.CANCELLED.get() > cancelled); From c0336588ed6efd8c1270e56f82e123ca3eac0770 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:16:13 -0400 Subject: [PATCH 3/3] fix: harden the GraphQL WebSocket client subscriptions Track subscribers from onSubscribe() so that stop() fails them all, signal non-positive demand without waiting for a result and complete the server operation, close the connection when its reader fails, send subscribe before any complete of a cancellation, signal setup failures, and count buffered results until they are delivered. Co-Authored-By: Claude Opus 5.5 --- .../graphql/GraphqlWebSocketTransport.java | 195 +++++++++++++----- .../GraphqlWebSocketClientTestCase.java | 52 +++++ 2 files changed, 195 insertions(+), 52 deletions(-) diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.java index 848ca25d74..ae0914901a 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketTransport.java @@ -18,6 +18,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; @@ -54,8 +55,12 @@ final class GraphqlWebSocketTransport implements GraphqlSubscriptionTransport { private final AtomicLong ids = new AtomicLong(); + /** The operations registered under their wire ID, from subscribe to completion. */ private final Map operations = new ConcurrentHashMap<>(); + /** The operations of all subscribers, from onSubscribe until terminated or cancelled. */ + private final Set subscriptions = ConcurrentHashMap.newKeySet(); + private Connection connection; private boolean stopped; @@ -82,10 +87,24 @@ public Publisher subscribe( } return subscriber -> { Objects.requireNonNull(subscriber, "subscriber"); - subscriber.onSubscribe(new Operation(subscriber, payload)); + Operation operation = new Operation(subscriber, payload); + subscriptions.add(operation); + subscriber.onSubscribe(operation); + // Checked once tracked, so that a concurrent stop either sees it or is seen here. + if (isStopped()) { + operation.stop(stoppedError()); + } }; } + private synchronized boolean isStopped() { + return stopped; + } + + private static GraphqlTransportException stoppedError() { + return new GraphqlTransportException("The GraphQL client was stopped", null, null, null); + } + @Override public void stop() { Connection current; @@ -94,10 +113,9 @@ public void stop() { current = connection; connection = null; } - GraphqlTransportException error = - new GraphqlTransportException("The GraphQL client was stopped", null, null, null); - for (Operation operation : List.copyOf(operations.values())) { - operation.fail(error); + GraphqlTransportException error = stoppedError(); + for (Operation operation : List.copyOf(subscriptions)) { + operation.stop(error); } if (current != null) { current.close(WebSocketSession.NORMAL_CLOSURE, null); @@ -108,9 +126,9 @@ public void stop() { /** Returns the open connection, opening one if needed. */ private synchronized Connection connection() throws GraphqlTransportException { if (stopped) { - throw new GraphqlTransportException("The GraphQL client was stopped", null, null, null); + throw stoppedError(); } - if (connection == null || !connection.session.isOpen()) { + if (connection == null || connection.closed || !connection.session.isOpen()) { connection = open(); } return connection; @@ -177,6 +195,9 @@ private final class Connection { private final Object sendLock = new Object(); + /** Set once the reader stops, after which the connection is no longer used. */ + private volatile boolean closed; + Connection(WebSocketSession session) { this.session = session; } @@ -196,6 +217,7 @@ void close(int code, String reason) { } void read() { + int closeCode = WebSocketSession.NORMAL_CLOSURE; try { WebSocketMessage received; while ((received = session.receive()) != null) { @@ -203,9 +225,15 @@ void read() { dispatch(received.getText()); } } - } catch (IOException | RuntimeException _) { + } catch (IOException _) { // Reported below as a closed connection + } catch (RuntimeException _) { + // A message that can't be decoded + closeCode = GraphqlWebSocketSupport.INVALID_MESSAGE; } + // Without a reader, the session can't be used anymore. + closed = true; + close(closeCode, null); GraphqlTransportException closed = new GraphqlTransportException( "The GraphQL WebSocket closed: " @@ -278,7 +306,7 @@ private final class Operation implements Subscription { private boolean started; - private boolean invalidDemand; + private volatile boolean invalidDemand; private volatile boolean cancelled; @@ -293,22 +321,35 @@ private final class Operation implements Subscription { @Override public void request(long n) { + boolean signal = false; + boolean start = false; synchronized (this) { if (cancelled) { return; } if (n <= 0) { invalidDemand = true; + // Without a delivery 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; - } - notifyAll(); - if (started) { - return; + start = !started; } started = true; + notifyAll(); + } + if (signal) { + subscriptions.remove(this); + subscriber.onError(invalidDemandError()); + } else if (start) { + Thread.ofVirtual().name("restlet-graphql-ws-operation").start(this::run); + } else if (n <= 0) { + // Wakes the delivery thread up, even without a result to deliver. + inbox.add(new Signal(null, 0, false, null)); } - Thread.ofVirtual().name("restlet-graphql-ws-operation").start(this::run); } @Override @@ -320,20 +361,46 @@ public void cancel() { cancelled = true; notifyAll(); } + subscriptions.remove(this); inbox.add(new Signal(null, 0, true, null)); - if (operations.remove(id, this) && !terminated) { - Connection current = connection; - if (current != null) { - try { - current.send("{\"id\":\"" + id + "\",\"type\":\"complete\"}"); - } catch (IOException _) { - // The connection is closing - } + completeOnServer(); + } + + /** Stops the operation because the transport stops, signalling an error. */ + void stop(Throwable error) { + boolean signal; + synchronized (this) { + if (cancelled) { + return; + } + // Without a delivery thread, signal from here and prevent one from starting. + signal = !started; + started = true; + if (signal) { + cancelled = true; } } + if (signal) { + subscriptions.remove(this); + subscriber.onError(error); + } else { + fail(error); + } } - void next(ExecutionResult result, long size) throws IOException { + /** Tells the server to complete the operation, if it is still active there. */ + private void completeOnServer() { + Connection current = connection; + if (operations.remove(id, this) && !terminated && current != null) { + try { + current.send("{\"id\":\"" + id + "\",\"type\":\"complete\"}"); + } catch (IOException _) { + // The connection is closing + } + } + } + + void next(ExecutionResult result, long size) { synchronized (this) { bufferedBytes += size; if (bufferedBytes <= adapter.getMaxResponseBytes()) { @@ -341,8 +408,6 @@ void next(ExecutionResult result, long size) throws IOException { return; } } - // Still registered: tell the server to complete the subscription. - boolean registered = operations.remove(id, this); fail( new GraphqlTransportException( "The subscription exceeded " @@ -351,9 +416,6 @@ void next(ExecutionResult result, long size) throws IOException { null, null, null)); - if (registered) { - connection.send("{\"id\":\"" + id + "\",\"type\":\"complete\"}"); - } } void error(ExecutionResult result) { @@ -368,10 +430,13 @@ void complete() { inbox.add(new Signal(null, 0, true, null)); } - /** Fails the operation at once, even ahead of buffered results. */ + /** + * Fails the operation at once, even ahead of buffered results. An operation still active on + * the server is completed there. + */ void fail(Throwable error) { + completeOnServer(); terminated = true; - operations.remove(id, this); synchronized (this) { if (failure == null) { failure = error; @@ -382,42 +447,60 @@ void fail(Throwable error) { } private void run() { + try { + if (subscribe()) { + deliver(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + subscriptions.remove(this); + } + } + + /** Sends the subscribe message. Returns false if the operation ended meanwhile. */ + private boolean subscribe() { try { Connection current = connection(); connection = current; - operations.put(id, this); - if (cancelled) { - operations.remove(id, this); - return; - } Map subscribe = new LinkedHashMap<>(); subscribe.put("id", id); subscribe.put("type", "subscribe"); subscribe.put("payload", payload); - current.send(GraphqlJson.writeString(subscribe)); - deliver(); + String text = GraphqlJson.writeString(subscribe); + // Sent while holding the lock that cancel() takes first, so that the complete + // message of a cancellation can't be sent before. + synchronized (this) { + if (cancelled) { + return false; + } + operations.put(id, this); + current.send(text); + } + return true; } catch (GraphqlTransportException e) { operations.remove(id, this); signalError(e); - } catch (IOException e) { + } catch (IOException | RuntimeException e) { + // Including failures of the request customizer or of the serialization operations.remove(id, this); signalError( new GraphqlTransportException( "Unable to send the GraphQL subscription", null, null, e)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); } + return false; } private void deliver() throws InterruptedException { while (true) { Signal signal = inbox.take(); - synchronized (this) { - bufferedBytes -= signal.size(); - } if (cancelled) { return; } + if (invalidDemand) { + rejectDemand(); + return; + } if (signal.error() != null) { signalError(signal.error()); return; @@ -427,9 +510,15 @@ private void deliver() throws InterruptedException { Throwable error = failure(); if (error != null) { signalError(error); + } else if (invalidDemand) { + rejectDemand(); } return; } + // Counted as buffered until delivered. + synchronized (this) { + bufferedBytes -= signal.size(); + } subscriber.onNext(signal.result()); } if (signal.last()) { @@ -439,6 +528,15 @@ private void deliver() throws InterruptedException { } } + private void rejectDemand() { + completeOnServer(); + signalError(invalidDemandError()); + } + + private IllegalArgumentException invalidDemandError() { + return new IllegalArgumentException("The requested number of results must be positive"); + } + private void signalError(Throwable error) { if (!cancelled) { cancelled = true; @@ -450,20 +548,13 @@ private synchronized Throwable failure() { return failure; } - /** Waits for demand. Returns false if cancelled or failed meanwhile. */ + /** Waits for demand. Returns false if cancelled, failed or given invalid demand. */ private boolean awaitDemand() throws InterruptedException { synchronized (this) { while (!cancelled && failure == null && demand == 0 && !invalidDemand) { wait(); } - if (cancelled || failure != null) { - return false; - } - if (invalidDemand) { - cancelled = true; - subscriber.onError( - new IllegalArgumentException( - "The requested number of results must be positive")); + if (cancelled || failure != null || invalidDemand) { return false; } if (demand != Long.MAX_VALUE) { diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java index 3378795e13..0224e5f678 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketClientTestCase.java @@ -219,6 +219,58 @@ void endpointsWithoutWebSocketFail() throws Exception { graphql.stop(); } + @Test + void stopFailsSubscribersWithoutDemand() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(subscriber); + graphql.stop(); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, subscriber.error.get()); + assertEquals(0, upgrades.get()); + } + + @Test + void nonPositiveDemandFailsAndCompletesTheServerSubscription() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.start(); + TestSubscriber idle = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(idle); + idle.subscription.request(0); + assertTrue(idle.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, idle.error.get()); + assertEquals(0, upgrades.get()); + + TestSubscriber active = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(active); + active.subscription.request(1); + await(() -> server.getActiveSubscriptions() == 1); + active.subscription.request(-1); + assertTrue(active.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, active.error.get()); + await(() -> server.getActiveSubscriptions() == 0); + graphql.stop(); + } + + @Test + void setupFailuresAreSignalled() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + graphql.setRequestCustomizer( + request -> { + throw new IllegalStateException("No credentials"); + }); + graphql.start(); + TestSubscriber subscriber = new TestSubscriber(); + graphql.subscribe("subscription { idle }", null, null).subscribe(subscriber); + subscriber.subscription.request(1); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + GraphqlTransportException e = + assertInstanceOf(GraphqlTransportException.class, subscriber.error.get()); + assertInstanceOf(IllegalStateException.class, e.getCause()); + graphql.stop(); + } + @Test void webSocketNeedsAStartedNettyClient() { GraphqlClientAdapter graphql = graphql("/graphql");