graphql-subscriptions

GraphQL subscriptions is a simple npm package that lets you wire up GraphQL with a pubsub system (like Redis) to implement subscriptions in GraphQL.

You can use it with any GraphQL client and server (not only Apollo).

Installation

npm install graphql-subscriptions graphql or yarn add graphql-subscriptions graphql

This package should be used with a network transport, for example subscriptions-transport-ws.

TypeScript

If you are developing a project that uses this module with TypeScript:

• ensure that your tsconfig.json lib definition includes "es2018.asynciterable"
• npm install @types/graphql or yarn add @types/graphql

Getting started with your first subscription

To begin with GraphQL subscriptions, start by defining a GraphQL Subscription type in your schema:

1type Subscription {
2  somethingChanged: Result
3}
4
5type Result {
6  id: String
7}

Next, add the Subscription type to your schema definition:

1schema {
2  query: Query
3  mutation: Mutation
4  subscription: Subscription
5}

Now, let's create a simple PubSub instance - it is a simple pubsub implementation, based on EventEmitter. Alternative EventEmitter implementations can be passed by an options object to the PubSub constructor.

1import { PubSub } from "graphql-subscriptions";
2
3export const pubsub = new PubSub();

If you're using TypeScript you can use the optional generic parameter for added type-safety:

1import { PubSub } from "graphql-subscriptions";
2
3const pubsub = new PubSub<{
4  EVENT_ONE: { data: number };
5  EVENT_TWO: { data: string };
6}>();
7
8pubsub.publish("EVENT_ONE", { data: 42 });
9pubsub.publish("EVENTONE", { data: 42 }); // ! ERROR
10pubsub.publish("EVENT_ONE", { data: "42" }); // ! ERROR
11pubsub.publish("EVENT_TWO", { data: "hello" });
12
13pubsub.subscribe("EVENT_ONE", () => {});
14pubsub.subscribe("EVENTONE", () => {}); // ! ERROR
15pubsub.subscribe("EVENT_TWO", () => {});

Next implement your Subscriptions type resolver using the pubsub.asyncIterableIterator to map the event you need:

1const SOMETHING_CHANGED_TOPIC = "something_changed";
2
3export const resolvers = {
4  Subscription: {
5    somethingChanged: {
6      subscribe: () => pubsub.asyncIterableIterator(SOMETHING_CHANGED_TOPIC),
7    },
8  },
9};

Subscriptions resolvers are not a function, but an object with subscribe method, that returns AsyncIterable.

The GraphQL engine now knows that somethingChanged is a subscription, and every time we use pubsub.publish it will publish content using our chosen transport layer:

1pubsub.publish(SOMETHING_CHANGED_TOPIC, { somethingChanged: { id: "123" } });

Note that the default PubSub implementation is intended for demo purposes. It only works if you have a single instance of your server and doesn't scale beyond a couple of connections. For production usage you'll want to use one of the PubSub implementations backed by an external store. (e.g. Redis)

Filters

When publishing data to subscribers, we need to make sure that each subscriber gets only the data it needs.

To do so, we can use withFilter helper from this package, which wraps AsyncIterator with a filter function, and lets you control each publication for each user.

withFilter API:

• asyncIteratorFn: (rootValue, args, context, info) => AsyncIterator<any> : A function that returns AsyncIterator you got from your pubsub.asyncIterableIterator.
• filterFn: (payload, variables, context, info) => boolean | Promise<boolean> - A filter function, executed with the payload (the published value), variables, context and operation info, must return boolean or Promise<boolean> indicating if the payload should pass to the subscriber.

For example, if somethingChanged would also accept a variable with the ID that is relevant, we can use the following code to filter according to it:

1import { withFilter } from "graphql-subscriptions";
2
3const SOMETHING_CHANGED_TOPIC = "something_changed";
4
5export const resolvers = {
6  Subscription: {
7    somethingChanged: {
8      subscribe: withFilter(
9        () => pubsub.asyncIterableIterator(SOMETHING_CHANGED_TOPIC),
10        (payload, variables) => {
11          return payload.somethingChanged.id === variables.relevantId;
12        }
13      ),
14    },
15  },
16};

Note that when using withFilter, you don't need to wrap your return value with a function.

Channels Mapping

You can map multiple channels into the same subscription, for example when there are multiple events that trigger the same subscription in the GraphQL engine.

1const SOMETHING_UPDATED = "something_updated";
2const SOMETHING_CREATED = "something_created";
3const SOMETHING_REMOVED = "something_removed";
4
5export const resolvers = {
6  Subscription: {
7    somethingChanged: {
8      subscribe: () =>
9        pubsub.asyncIterableIterator([
10          SOMETHING_UPDATED,
11          SOMETHING_CREATED,
12          SOMETHING_REMOVED,
13        ]),
14    },
15  },
16};

Payload Manipulation

You can also manipulate the published payload, by adding resolve methods to your subscription:

1const SOMETHING_UPDATED = "something_updated";
2
3export const resolvers = {
4  Subscription: {
5    somethingChanged: {
6      resolve: (payload, args, context, info) => {
7        // Manipulate and return the new value
8        return payload.somethingChanged;
9      },
10      subscribe: () => pubsub.asyncIterableIterator(SOMETHING_UPDATED),
11    },
12  },
13};

Note that resolve methods execute after subscribe, so if the code in subscribe depends on a manipulated payload field, you will need to factor out the manipulation and call it from both subscribe and resolve.

Usage with callback listeners

Your database might have callback-based listeners for changes, for example something like this:

1const listenToNewMessages = (callback) => {
2  return db.table("messages").listen((newMessage) => callback(newMessage));
3};
4
5// Kick off the listener
6listenToNewMessages((message) => {
7  console.log(message);
8});

The callback function would be called every time a new message is saved in the database. Unfortunately, that doesn't play very well with async iterators out of the box because callbacks are push-based, where async iterators are pull-based.

We recommend using the callback-to-async-iterator module to convert your callback-based listener into an async iterator:

1import asyncify from "callback-to-async-iterator";
2
3export const resolvers = {
4  Subscription: {
5    somethingChanged: {
6      subscribe: () => asyncify(listenToNewMessages),
7    },
8  },
9};

Custom AsyncIterator Wrappers

The value you should return from your subscribe resolver must be an AsyncIterable.

You can wrap an AsyncIterator with custom logic for your subscriptions. For compatibility with APIs that require AsyncIterator or AsyncIterable, your wrapper can return an AsyncIterableIterator to comply with both.

For example, the following implementation manipulates the payload by adding some static fields:

1export const withStaticFields = (
2  asyncIterator: AsyncIterator<any>,
3  staticFields: Object
4): Function => {
5  return (
6    rootValue: any,
7    args: any,
8    context: any,
9    info: any
10  ): AsyncIterableIterator<any> => {
11    return {
12      next() {
13        return asyncIterator.next().then(({ value, done }) => {
14          return {
15            value: {
16              ...value,
17              ...staticFields,
18            },
19            done,
20          };
21        });
22      },
23      return() {
24        return Promise.resolve({ value: undefined, done: true });
25      },
26      throw(error) {
27        return Promise.reject(error);
28      },
29      [Symbol.asyncIterator]() {
30        return this;
31      },
32    };
33  };
34};

You can also take a look at withFilter for inspiration.

PubSub Implementations

It can be easily replaced with some other implementations of PubSubEngine abstract class. Here are a few of them:

• Use Redis with https://github.com/davidyaha/graphql-redis-subscriptions
• Use Google PubSub with https://github.com/axelspringer/graphql-google-pubsub
• Use MQTT enabled broker with https://github.com/aerogear/graphql-mqtt-subscriptions
• Use RabbitMQ with https://github.com/cdmbase/graphql-rabbitmq-subscriptions
• Use AMQP (RabbitMQ) with https://github.com/Surnet/graphql-amqp-subscriptions
• Use Kafka with https://github.com/ancashoria/graphql-kafka-subscriptions
• Use Kafka (using Kafkajs) with https://github.com/tomasAlabes/graphql-kafkajs-subscriptions
• Use Postgres with https://github.com/GraphQLCollege/graphql-postgres-subscriptions
• Use NATS with https://github.com/moonwalker/graphql-nats-subscriptions
• Use Mongoose (MongoDB) with https://github.com/Nickolasmv/graphql-mongoose-subscriptions
• Use multiple backends with https://github.com/jcoreio/graphql-multiplex-subscriptions
• Use Ably for multi-protocol support with https://github.com/ably-labs/graphql-ably-pubsub
• Use Google Firestore with https://github.com/m19c/graphql-firestore-subscriptions
• Use Amazon's Simple Notification Service (SNS) and Simple Queue Service (SQS) with https://github.com/sagahead-io/graphql-snssqs-subscriptions
• Add your implementation...

You can also implement a PubSub of your own, by using the exported abstract class PubSubEngine from this package. By using extends PubSubEngine you use the default asyncIterator method implementation; by using implements PubSubEngine you must implement your own AsyncIterator.

SubscriptionManager @deprecated

SubscriptionManager is the previous alternative for using graphql-js subscriptions directly, and it's now deprecated.

If you are looking for its API docs, refer to a previous commit of the repository