/lib/graphql-rx
Contents
This optional module provides the publisher and subscriber objects used by GraphQL subscriptions.
Subscription events need a transport that stays open, and the library does not provide one. A universal API can serve either a WebSocket channel or an SSE stream, and each event is an ExecutionResult either way.
The examples below use SSE, the shorter of the two: it needs no handshake, and a browser consumes it with EventSource.
Choose a WebSocket instead when the consumer is a standard GraphQL client. GraphiQL, Apollo Client and similar tools speak the graphql-transport-ws protocol, which expects the server to answer connection_init with connection_ack and to wrap each event in a next message carrying the operation’s id. An API serves that from a webSocketEvent handler, the counterpart to the sseEvent handler used below. The publisher and subscriber on this page behave identically under either transport.
Functions
createPublishProcessor
Creates a PublishProcessor that can be returned by a subscription field resolver.
Returns
PublishProcessor — a reactive event source.
Example
Create the processor outside the resolver, so the same instance survives across requests and can be published to from anywhere in the application.
const graphQlRxLib = require('/lib/graphql-rx');
const personProcessor = graphQlRxLib.createPublishProcessor();
const rootSubscriptionType = schemaGenerator.createObjectType({
name: 'Subscription',
fields: {
personAdded: {
type: personType,
resolve: () => personProcessor
}
}
});createSubscriber
Creates a subscriber for the publisher returned as data by a subscription execution.
Parameters
createSubscriber() takes a single params object with these properties:
| Name | Type | Description |
|---|---|---|
|
|
function |
Optional. Called with each mapped ExecutionResult produced by a subscription. |
Returns
Example
Executing a subscription returns a publisher as data rather than a result object. Subscribing to it starts delivery, and each event arriving in onNext is a full ExecutionResult.
Send to a named group rather than to a single clientId. Connection ids must not be collected in module-scope state, a group name is a constant, and group membership is cleaned up automatically when a client disconnects.
const sseLib = require('/lib/xp/sse');
exports.GET = () => ({
sse: {
retry: 5000
}
});
exports.sseEvent = (event) => {
if (event.type === 'open') {
sseLib.addToGroup({group: 'people', clientId: event.clientId});
}
};const graphQlLib = require('/lib/graphql');
const graphQlRxLib = require('/lib/graphql-rx');
const sseLib = require('/lib/xp/sse');
const result = graphQlLib.execute(
schema,
'subscription { personAdded { name } }'
);
const subscriber = graphQlRxLib.createSubscriber({
onNext: (event) => sseLib.sendToGroup({
group: 'people',
message: {
event: 'personAdded',
data: JSON.stringify(event)
}
})
});
result.data.subscribe(subscriber);Classes
PublishProcessor
A reactive event source returned by createPublishProcessor(). It has no constructor.
All the examples below use a processor created with createPublishProcessor():
const graphQlRxLib = require('/lib/graphql-rx');
const personProcessor = graphQlRxLib.createPublishProcessor();onNext
Publishes a value to active subscribers.
Parameters
| Name | Type | Description |
|---|---|---|
|
|
any |
Value delivered to every active subscriber. Becomes the |
Example
The published value becomes the source of the subscription field’s resolver, so it is the application’s own data, not an execution result.
const peopleLib = require('/lib/people');
personProcessor.onNext(peopleLib.getPerson(name));onError
Terminates the stream with an error.
Parameters
| Name | Type | Description |
|---|---|---|
|
|
Java |
Error that terminates the stream. Construct it with |
Example
The argument is a Java Throwable, so construct one with Java.type() rather than passing a string or a JavaScript Error.
const IllegalStateException = Java.type('java.lang.IllegalStateException');
personProcessor.onError(new IllegalStateException('Person event source became unavailable'));onComplete
Completes the stream.
Example
Subscribers receive no further events after this call.
personProcessor.onComplete();filter
Forwards only the values a predicate accepts.
Parameters
| Name | Type | Description |
|---|---|---|
|
|
function |
Called with each published value. The value is forwarded when the return value is truthy. |
Returns
A publisher carrying only the values the predicate accepted. It exposes subscribe() and filter(), so filters can be chained.
Example
Filtering returns a new publisher and leaves the processor untouched, so one processor can back several subscription fields with different predicates.
const rootSubscriptionType = schemaGenerator.createObjectType({
name: 'Subscription',
fields: {
personAdded: {
type: personType,
resolve: () => personProcessor.filter(
(person) => person.age >= 18
)
}
}
});subscribe
Subscribes a compatible reactive-streams subscriber.
Parameters
| Name | Type | Description |
|---|---|---|
|
|
Subscriber that receives the published values, created with createSubscriber(). |
Example
Application code normally reaches the publisher as result.data after executing a subscription, and calls subscribe() on that. Calling it on the processor directly is equivalent when the resolver returns the processor unfiltered.
const subscriber = graphQlRxLib.createSubscriber({
onNext: (event) => sseLib.sendToGroup({
group: 'people',
message: {data: JSON.stringify(event)}
})
});
personProcessor.subscribe(subscriber);SubscriptionSubscriber
A subscriber returned by createSubscriber(). It has no constructor.
cancelSubscription
Cancels its active subscription. Calling it before subscription or more than once has no effect.
Example
Cancel from wherever the subscriber is held, so the processor stops delivering to a client that has gone away. For a subscription serving one client, the SSE close event is the reliable trigger: it is terminal, firing after timeout and after error as well. A subscription feeding a broadcast group outlives any single client and is not cancelled here.
exports.sseEvent = (event) => {
if (event.type === 'close') {
subscriber.cancelSubscription();
}
};