Subscriptions
Contents
Stream events to clients as they happen.
A subscription differs from a query in what the engine gives back. Where a query resolves to a value, a subscription resolves to a publisher that emits a result every time something happens. GraphQL itself says nothing about how those results reach the client — that is the transport’s job, and XP offers two: SSE and WebSocket.
We will use SSE. The subscription is sent to the same endpoint as everything else, and the response is a stream that stays open instead of a JSON body — no handshake, and it can be watched with cURL.
|
This chapter shows how subscriptions wire into XP. It is not a production blueprint.
The schema side carries over to either transport: a subscription resolves to a publisher, the publisher is shared and its listener registered once in |
To achieve this, we need to:
-
Add the libraries for events and SSE that ship with Enonic XP.
-
Create a GraphQL Subscription type and add it to our GraphQL schema.
-
Answer a subscription with a stream instead of a JSON body.
Root Subscription GraphQL type
Add lib-event and lib-sse. Both ship with XP, so they come from the xplibs catalog and carry no version of their own.
include xplibs.event
include xplibs.sseA Subscription type is built like any other object type. Ours has one field, event, typed as Json.
const rootSubscriptionType = schemaGenerator.createObjectType({
name: 'Subscription',
fields: {
event: {
type: Json,
resolve: () => {
// Implementation
},
}
}
});Assign it to subscription when creating the schema:
const graphQLSchema = schemaGenerator.createSchema({
query: rootQueryType,
mutation: rootMutationType,
subscription: rootSubscriptionType,
dictionary: [noteType]
});That leaves the resolve function.
The library’s /lib/graphql-rx module supplies the two pieces a subscription needs: createPublishProcessor, an event source a resolver can return, and createSubscriber, which receives what the processor publishes.
import {createPublishProcessor} from '/lib/graphql-rx';And lib-event, to listen for the events themselves:
import {listener, send} from '/lib/xp/event';The resolver’s job is to return a processor that publishes the events we care about. Two things have to be true for that to work: the processor must be the same object for every subscriber, and the event listener that feeds it must be registered exactly once.
| Neither belongs in the resolver. A resolver runs on every subscribe, so a processor created there would be private to one client, and a listener registered there would be added again for every client and never removed. |
Put the processor in a module of its own, so everything that needs it shares one instance:
import {createPublishProcessor} from '/lib/graphql-rx';
export const noteProcessor = createPublishProcessor();Register the listener in main.ts, the script XP runs when the application starts:
import {listener} from '/lib/xp/event';
import {noteProcessor} from '/lib/events';
export function init(): void {
listener({
type: 'custom.note.*',
callback: (event) => {
noteProcessor.onNext(event);
}
});
}
init();This is where a listener belongs. main.ts runs once as the application loads, so the listener is registered once, rather than on the first request that happens to reach a controller.
The resolver then just hands out the shared processor:
import {noteProcessor} from '/lib/events';
const rootSubscriptionType = schemaGenerator.createObjectType({
name: 'Subscription',
fields: {
event: {
type: Json,
resolve: () => noteProcessor,
}
}
});A single processor serves every subscriber. When two subscription fields need different slices of the same stream, call filter() on the processor instead of creating another one.
Nothing publishes yet. Send an event from the mutation resolvers — note.created in createNote, and note.deleted in deleteNote:
send({
type: 'note.created',
distributed: true,
data: {
note: note,
}
});Stream the results
Our API already exports POST, and a subscription arrives on it like any other operation. The difference is what the engine returns: a query or a mutation resolves to data that can be serialized as JSON, while a subscription resolves to a publisher that emits results over time. So the controller decides how to answer by looking at the operation:
import {send} from '/lib/xp/sse';
interface StreamAttributes {
query: string;
variables?: unknown;
}
const SUBSCRIPTION = /^\s*subscription\b/; (1)
export function POST(req: Request): Response {
const body = JSON.parse(req.body as string) as {query?: string; variables?: unknown};
if (!body.query) {
return badRequest('Missing `query` in request body.');
}
if (SUBSCRIPTION.test(body.query)) {
return {
sse: { (2)
attributes: body.variables === undefined (3)
? {query: body.query}
: {query: body.query, variables: body.variables},
retry: 5000, (4)
}
};
}
const result = execute(graphQLSchema, body.query, body.variables);
return {
contentType: 'application/json',
body: result,
};
}| 1 | Decided before anything is executed, so the endpoint never runs an operation just to discover what kind it was. | ||
| 2 | Returning an sse object turns the response into an open stream. This is the same shape a GET handler would return — SSE is not tied to GET in XP. |
||
| 3 | Per-connection state that XP hands back on every event for this connection. The sseEvent handler receives no request, so this is how the subscription document reaches it; keeping a map of connections would be the alternative, and worse.
|
||
| 4 | How long the client should wait before reconnecting, in milliseconds. |
Opening the stream does not run the subscription. That happens when XP fires the open event, which is also where we get the clientId needed to send anything back:
const subscribers: Record<string, Subscriber> = {};
export function sseEvent(event: SseEvent<StreamAttributes>): void {
if (event.type === 'open') {
openSubscription(event.clientId, event.attributes as StreamAttributes);
return;
}
if (event.type === 'close') { (1)
cancelSubscription(event.clientId);
}
}
function isPublisher(data: unknown): data is Publisher {
return !!data && typeof (data as Publisher).subscribe === 'function';
}
function openSubscription(clientId: string, attributes: StreamAttributes): void {
const result = execute(graphQLSchema, attributes.query, attributes.variables);
if (!isPublisher(result.data)) { (2)
send({
clientId: clientId,
message: {
event: 'error',
data: JSON.stringify({errors: result.errors ?? [{message: 'The document is not a subscription.'}]})
}
});
return;
}
const subscriber = createSubscriber({
onNext: (executionResult) => { (3)
send({
clientId: clientId,
message: {event: 'next', data: JSON.stringify(executionResult)}
});
}
});
subscribers[clientId] = subscriber;
result.data.subscribe(subscriber);
}
function cancelSubscription(clientId: string): void {
const subscriber = subscribers[clientId];
if (subscriber) {
delete subscribers[clientId];
subscriber.cancelSubscription();
}
}| 1 | close is terminal and fires after timeout and error as well, so it is the one place a subscription needs cancelling. Without it, a client that simply goes away leaves the processor delivering to nobody. |
| 2 | A document can start with subscription and still fail — a misspelled field, for example. Reporting that as an error frame is friendlier than an empty stream that never delivers. |
| 3 | Each event is a complete execution result — the same shape a query returns — so a client parses it the same way. |
Test the subscription
Open the stream in one terminal. cURL’s -N disables buffering, so events appear as they arrive rather than when the connection ends:
curl -N -X POST "http://localhost:8080/api/com.example.gqltutorial:graphql" \
-H "Content-Type: application/json" \
-d '{"query":"subscription { event }"}'The connection stays open with no output yet. In a second terminal, create a note:
curl -X POST "http://localhost:8080/api/com.example.gqltutorial:graphql" \
-H "Content-Type: application/json" \
-d '{"query":"mutation { createNote(title: \"My Note\", content: \"Brief content\") { id } }"}'The first terminal receives an event:
retry:5000
event:next
data:{"data":{"event":{"type":"custom.note.created","timestamp":1752491906282,"localOrigin":true,"distributed":true,"data":{"note":{"id":"2g6inpt0ybexu","title":"My Note","content":"Brief content","createdAt":"2026-08-13T11:18:26.281Z"}}}}}Delete a note and a second event arrives. Stop the stream with Ctrl+C; the close event fires and the subscription is cancelled server-side.
As noted at the top of this chapter, a standard GraphQL client will subscribe over a WebSocket instead — one connection can then carry several operations chosen by the client. The /lib/graphql-rx page of the GraphQL Library documentation covers that variant. |
Where to go next
You now have a complete GraphQL API: a schema with queries, mutations and subscriptions, served over HTTP from a universal API.
Before building anything real on top of it, revisit the shortcuts this tutorial took:
-
Notes are kept in memory. Persist them with lib-content or lib-node.
-
The API allows
role:system.everyone. Narrow theallowlist ingraphql.yaml, and authorize inside the resolvers using the context passed toexecute(). -
Only a small part of the library is used here. The GraphQL Library documentation also covers interfaces, unions, enums, input types and Relay-style connections.