Repository navigation
Conversation
graphql-sse in distinct connections mode, on core SSE: - Server: GraphqlTransport.SSE enables event streams for requests accepting text/event-stream: subscriptions (also over GET when it is enabled), and queries and mutations as one next event. Results are requested one at a time and written as next events, then complete. Request errors are 400 with a JSON body. An initial comment sends the response head at once; heartbeats (15 s by default) keep idle streams open and detect disconnected clients. A failed write cancels the subscription. - Limits: 64 active subscriptions per adapter (503 beyond), 1 MiB of queued bytes per subscription (the stream ends with an error and the subscription is cancelled). Stopping the adapter ends its open streams so that the server connector can shut down gracefully. - Client: subscribe() returns a cold Publisher<ExecutionResult>; each subscriber sends its own request on its first demand, from a virtual thread that emits all the signals. Demand is honored, cancel() aborts the request, events are bounded by maxResponseBytes, request errors are published as a single result, and a stream ending without a complete event is a GraphqlTransportException. stop() fails the open subscriptions. Refs #1585 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
|
This was referenced Oct 30, 2026
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
Early client disconnects can bypass cleanup and permanently exhaust subscription capacity, with additional unresolved subscription-lifecycle and SSE protocol defects.
8 open issues
What changed in this PR
Adds GraphQL subscriptions over Server-Sent Events (SSE), extending the existing HTTP binding with graphql-sse distinct-connections support built on Restlet’s core SSE APIs.
Changes:
- Adds server streaming with heartbeats, subscription limits, back-pressure, and shutdown handling.
- Adds cold client publishers with demand handling, cancellation, and transport-error reporting.
- Adds SSE integration tests, idle-publisher fixtures, and a changelog entry.
| File | Description |
|---|---|
org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/TickPublisher.java |
Adds idle-publisher behavior for tests. |
org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlTestSchemas.java |
Adds an idle subscription field. |
org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlSseTestCase.java |
Tests SSE protocol, limits, demand, and lifecycle behavior. |
org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlEventStream.java |
Writes subscription results as bounded SSE streams. |
org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlTransport.java |
Defines the SSE subscription transport. |
org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlSsePublisher.java |
Implements client-side SSE subscriptions. |
org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java |
Integrates SSE negotiation, configuration, and stream lifecycle. |
org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java |
Exposes subscription APIs and tracks open subscriptions. |
changes.md |
Documents GraphQL SSE support and default limits. |
Review effort: Balanced
Comment on lines
+162
to
+167
| // An empty comment sends the response head at once, before the first result. | ||
| sink.send(OPEN); | ||
| QueueingSubscriber subscriber = new QueueingSubscriber(maxQueuedBytes); | ||
| this.subscriber = subscriber; | ||
| openStreams.add(this); | ||
| try { |
Comment on lines
+398
to
+401
| private static Publisher<ExecutionResult> single(ExecutionResult result) { | ||
| return subscriber -> | ||
| subscriber.onSubscribe( | ||
| new Subscription() { |
Comment on lines
+355
to
+356
| subscriptionTransports = | ||
| Collections.unmodifiableSet(EnumSet.copyOf(subscriptionTransports)); |
Comment on lines
+455
to
+457
| if (subscription && activeSubscriptions.get() >= maxSubscriptions) { | ||
| writeEntity(response, responseType, failure(error("Too many active subscriptions"))); | ||
| response.setStatus(Status.SERVER_ERROR_SERVICE_UNAVAILABLE); |
| * reported with {@code 400} in {@code application/graphql-response+json} only. | ||
| */ | ||
| private static void write(Response response, MediaType responseType, ExecutionResult result) { | ||
| write(response, responseType, result, false); |
Comment on lines
+58
to
+59
| subscriber.onSubscribe(subscription); | ||
| if (!adapter.register(subscription)) { |
Comment on lines
+91
to
+92
| if (n <= 0) { | ||
| invalidDemand = true; |
Comment on lines
+114
to
+116
| Request current = request; | ||
| if (current != null) { | ||
| current.abort(); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #1585. Part of #1581.
Summary
Subscriptions with
graphql-ssein distinct connections mode, on core SSE (EventStreamRepresentation,EventStreamReader). WebSocket comes in G4.Server
text/event-stream(named explicitly; wildcards don't select it) gets an event stream:nextevents carrying results, thencomplete. Queries and mutations are onenext; subscriptions are also accepted overGETwhenGETis enabled. Request errors are400with a JSON body, as ingraphql-sse.503beyond), 1 MiB of queued bytes per subscription. The queue only grows when a publisher ignores demand or a result is huge; beyond the limit the stream ends with an error and the subscription is cancelled. The connector'sserver.maxEventStreamsstill applies.stop()ends the adapter's open streams (error +complete), so that the server connector shuts down gracefully instead of waiting for idle subscriptions.Client
subscribe(String|Document, operationName, variables)returns a coldPublisher<ExecutionResult>. Each subscriber sends its ownPOSTon its first demand, from a virtual thread that emits all signals (no concurrentonNext).cancel()aborts the request (Request.abort());request(n <= 0)signalsIllegalArgumentException.400, or local validation withsetSchema) are published as a single result. Non-GraphQL responses, events abovemaxResponseBytesand streams ending withoutcompleteare aGraphqlTransportException.stop()fails open subscriptions with aGraphqlTransportException; the borrowed client keeps running.Core findings
EventStreamReaderhas no line or event size limit (BufferedReader.readLine). The client wraps the stream with a per-event byte budget (one 8 KiB read-ahead margin); a bounded reader in core would be better.EventStreamRepresentationdoes not commit the response head until the first event or heartbeat.Validation
./mvnw -B verify -P default,test -pl org.restlet.ext.graphql: 47 tests, 0 failures; 93.8% instruction coverage (gate 80%); Spotless, SpotBugs, Javadoc,-Werrorand dependency convergence pass.New
GraphqlSseTestCase(16 tests): wire format ofnext/complete; queries over SSE; JSON request errors;GETsubscriptions andGETmutations rejected; SSE off unless enabled; client disconnect cancels a busy subscription; heartbeats detect a disconnected idle client; subscription cap; queued bytes limit; client demand (no result without demand), completion, cancellation aborting the server-side subscription,stop(), request errors as results (remote and local validation), transport errors (HTML502, truncated stream, oversized event), invalid demand; configuration validation.🤖 Generated with Claude Code