diff --git a/changes.md b/changes.md index b86018b38d..1fdea34b88 100644 --- a/changes.md +++ b/changes.md @@ -152,7 +152,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..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 @@ -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,61 @@ public void handle(Request request, Response response) { } } - private ExecutionInput.Builder customize(Request request, ExecutionInput.Builder input) { + /** + * 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 execution input. + */ + ExecutionInput prepare( + Request request, ExecutionInput.Builder input, Map initPayload) { + 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( + 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..208ef4f64d --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlWebSocketSupport.java @@ -0,0 +1,501 @@ +/** + * 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.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +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; + + /** 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; + } + + @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() { + // 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)) { + 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) { + 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 final Object initLock = new Object(); + + private boolean initTimedOut; + + 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); + 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 { + 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 _) { + // Interrupted once the connection ends. + Thread.currentThread().interrupt(); + return; + } + synchronized (initLock) { + if (initialized) { + return; + } + initTimedOut = true; + } + 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"); + } + 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" -> { + 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; + } + } + } + + /** + * 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, closeReason(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() { + AtomicInteger active = adapter.getActiveSubscriptionCounter(); + boolean reserved = false; + try { + // 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()) { + 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 (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); + 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) { + 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 (!send( + "{\"id\":" + + GraphqlJson.writeString(id) + + ",\"type\":\"next\",\"payload\":" + + next.json() + + "}", + false)) { + return; + } + current.request(); + } else if (signal instanceof GraphqlResultQueue.Failure failure) { + if (failure.cause() != null) { + adapter.getLogger() + .log( + Level.WARNING, + "GraphQL subscription failed", + failure.cause()); + } + send(message("error", id, error(failure.message())), true); + return; + } else if (signal instanceof GraphqlResultQueue.Completed) { + send(message("complete", id, null), true); + return; + } else { + return; + } + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + } finally { + current.cancel(); + } + } + } +} 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..83345cec38 --- /dev/null +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlWebSocketServerTestCase.java @@ -0,0 +1,399 @@ +/** + * 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); + // Long enough for a cold or busy JVM to send connection_init. The 4408 test has its own. + adapter.setConnectionInitTimeout(Duration.ofSeconds(10)); + component.getDefaultHost().attach("/graphql", adapter); + component.start(); + base = "localhost:" + component.getServers().get(0).getActualPort() + "/graphql"; + http = HttpClient.newHttpClient(); + } + + @AfterEach + void tearDown() throws Exception { + // Not close(), which waits for every WebSocket, forever if a failed test left one open. + http.shutdownNow(); + component.stop(); + } + + private Peer connect() throws Exception { + return connect(base); + } + + private Peer connect(String target) throws Exception { + Peer peer = new Peer(); + peer.socket = + http.newWebSocketBuilder() + .subprotocols(GraphqlWebSocketSupport.SUBPROTOCOL) + .buildAsync(URI.create("ws://" + target), 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(); + int subscribed = TickPublisher.SUBSCRIBED.get(); + Peer peer = initialized(null); + peer.send(subscribe("a", "subscription { idle }")); + peer.send(subscribe("b", "subscription { idle }")); + // Slots are reserved before execution: wait for both publishers to be subscribed. + await(() -> TickPublisher.SUBSCRIBED.get() == subscribed + 2); + assertEquals(2, adapter.getActiveSubscriptions()); + 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()); + + GraphqlServerAdapter impatient = new GraphqlServerAdapter(); + impatient.setGraphQL(GraphqlTestSchemas.graphQL(new AtomicInteger())); + impatient.getSubscriptionTransports().add(GraphqlTransport.WEBSOCKET); + impatient.setConnectionInitTimeout(Duration.ofMillis(500)); + component.getDefaultHost().attach("/impatient", impatient); + impatient.start(); + Peer slow = connect(base.replace("/graphql", "/impatient")); + assertEquals(4408, slow.closeCode()); + impatient.stop(); + + 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 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()); + } + + int subscribed = TickPublisher.SUBSCRIBED.get(); + peer.send(subscribe("a", "subscription { idle }")); + await(() -> TickPublisher.SUBSCRIBED.get() == subscribed + 1); + assertEquals(1, limited.getActiveSubscriptions()); + 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(); + 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)); + } + } +} diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java index 3e4384f924..25a078ac0a 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java @@ -24,6 +24,9 @@ final class TickPublisher implements Publisher { static final AtomicInteger CANCELLED = new AtomicInteger(); + /** Counts subscriptions, which can start after the subscription slot is reserved. */ + static final AtomicInteger SUBSCRIBED = new AtomicInteger(); + private final int count; private final boolean idle; @@ -41,6 +44,7 @@ final class TickPublisher implements Publisher { @Override public void subscribe(Subscriber subscriber) { subscriber.onSubscribe(new Ticks(subscriber)); + SUBSCRIBED.incrementAndGet(); } private final class Ticks implements Subscription {