graphql-redis-subscriptions

This package implements the PubSubEngine Interface from the graphql-subscriptions package and also the new AsyncIterator interface. It allows you to connect your subscriptions manager to a Redis Pub Sub mechanism to support multiple subscription manager instances.

Installation

At first, install the graphql-redis-subscriptions package:

npm install graphql-redis-subscriptions

As the graphql-subscriptions package is declared as a peer dependency, you might receive warning about an unmet peer dependency if it's not installed already by one of your other packages. In that case you also need to install it too:

npm install graphql-subscriptions

Using as AsyncIterator

Define your GraphQL schema with a Subscription type:

1schema {
2  query: Query
3  mutation: Mutation
4  subscription: Subscription
5}
6
7type Subscription {
8    somethingChanged: Result
9}
10
11type Result {
12    id: String
13}

Now, let's create a simple RedisPubSub instance:

1import { RedisPubSub } from 'graphql-redis-subscriptions';
2const pubsub = new RedisPubSub();

Now, implement your Subscriptions type resolver, using the pubsub.asyncIterator to map the event you need:

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

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

Calling the method asyncIterator of the RedisPubSub instance will send redis a SUBSCRIBE message to the topic provided and will return an AsyncIterator binded to the RedisPubSub instance and listens to any event published on that topic. Now, the GraphQL engine knows that somethingChanged is a subscription, and every time we will use pubsub.publish over this topic, the RedisPubSub will PUBLISH the event over redis to all other subscribed instances and those in their turn will emit the event to GraphQL using the next callback given by the GraphQL engine.

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

Dynamically create a topic based on subscription args passed on the query

1export const resolvers = {
2  Subscription: {
3    somethingChanged: {
4      subscribe: (_, args) => pubsub.asyncIterator(`${SOMETHING_CHANGED_TOPIC}.${args.relevantId}`),
5    },
6  },
7}

Using a pattern on subscription

1export const resolvers = {
2  Subscription: {
3    somethingChanged: {
4      subscribe: (_, args) => pubsub.asyncIterator(`${SOMETHING_CHANGED_TOPIC}.${args.relevantId}.*`, { pattern: true })
5    },
6  },
7}

Using both arguments and payload to filter events

1import { withFilter } from 'graphql-subscriptions';
2
3export const resolvers = {
4  Subscription: {
5    somethingChanged: {
6      subscribe: withFilter(
7        (_, args) => pubsub.asyncIterator(`${SOMETHING_CHANGED_TOPIC}.${args.relevantId}`),
8        (payload, variables) => payload.somethingChanged.id === variables.relevantId,
9      ),
10    },
11  },
12}

Configuring RedisPubSub

RedisPubSub constructor can be passed a configuration object to enable some advanced features.

1export interface PubSubRedisOptions {
2  connection?: RedisOptions | string;
3  triggerTransform?: TriggerTransform;
4  connectionListener?: (err?: Error) => void;
5  publisher?: RedisClient;
6  subscriber?: RedisClient;
7  reviver?: Reviver;
8  serializer?: Serializer;
9  deserializer?: Deserializer;
10  messageEventName?: string;
11  pmessageEventName?: string;
12}

option	type	default	description
connection	object | string	undefined	the connection option is passed as is to the ioredis constructor to create redis subscriber and publisher instances. for greater controll, use publisher and subscriber options.
triggerTransform	function	(trigger) => trigger	deprecated
connectionListener	function	undefined	pass in connection listener to log errors or make sure connection to redis instance was created successfully.
publisher	function	undefined	must be passed along side subscriber. see #creating-a-redis-client
subscriber	function	undefined	must be passed along side publisher. see #creating-a-redis-client
reviver	function	undefined	see #using-a-custom-reviver
serializer	function	undefined	see #using-a-custom-serializerdeserializer
deserializer	function	undefined	see #using-a-custom-serializerdeserializer
messageEventName	string	undefined	see #receiving-messages-as-buffers
pmessageEventName	string	undefined	see #receiving-messages-as-buffers

Creating a Redis Client

The basic usage is great for development and you will be able to connect to a Redis server running on your system seamlessly. For production usage, it is recommended to pass a redis client (like ioredis) to the RedisPubSub constructor. This way you can control all the options of your redis connection, for example the connection retry strategy.

1import { RedisPubSub } from 'graphql-redis-subscriptions';
2import * as Redis from 'ioredis';
3
4const options = {
5  host: REDIS_DOMAIN_NAME,
6  port: PORT_NUMBER,
7  retryStrategy: times => {
8    // reconnect after
9    return Math.min(times * 50, 2000);
10  }
11};
12
13const pubsub = new RedisPubSub({
14  ...,
15  publisher: new Redis(options),
16  subscriber: new Redis(options)
17});

Receiving messages as Buffers

Some Redis use cases require receiving binary-safe data back from redis (in a Buffer). To accomplish this, override the event names for receiving messages and pmessages. Different redis clients use different names, for example:

library	message event	message event (Buffer)	pmessage event	pmessage event (Buffer)
ioredis	message	messageBuffer	pmessage	pmessageBuffer
node-redis	message	message_buffer	pmessage	pmessage_buffer

1import { RedisPubSub } from 'graphql-redis-subscriptions';
2import * as Redis from 'ioredis';
3
4const pubsub = new RedisPubSub({
5  ...,
6  // Tells RedisPubSub to register callbacks on the messageBuffer and pmessageBuffer EventEmitters
7  messageEventName: 'messageBuffer',
8  pmessageEventName: 'pmessageBuffer',
9});

Also works with your Redis Cluster

1import { RedisPubSub } from 'graphql-redis-subscriptions';
2import { Cluster } from 'ioredis';
3
4const cluster = new Cluster(REDIS_NODES); // like: [{host: 'ipOrHost', port: 1234}, ...]
5const pubsub = new RedisPubSub({
6  ...,
7  publisher: cluster,
8  subscriber: cluster
9});

You can learn more on the ioredis package here.

Using a custom serializer/deserializer

By default, Javascript objects are (de)serialized using the JSON.stringify and JSON.parse methods. You may pass your own serializer and/or deserializer function(s) as part of the options.

The deserializer will be called with an extra context object containing pattern (if available) and channel properties, allowing you to access this information when subscribing to a pattern.

1import { RedisPubSub } from 'graphql-redis-subscriptions';
2import { someSerializer, someDeserializer } from 'some-serializer-library';
3
4const serialize = (source) => {
5  return someSerializer(source);
6};
7
8const deserialize = (sourceOrBuffer, { channel, pattern }) => {
9  return someDeserializer(sourceOrBuffer, channel, pattern);
10};
11
12const pubSub = new RedisPubSub({ ..., serializer: serialize, deserializer: deserialize });

Using a custom reviver

By default, Javascript objects are serialized using the JSON.stringify and JSON.parse methods. This means that not all objects - such as Date or Regexp objects - will deserialize correctly without a custom reviver, that work out of the box with the default in-memory implementation. For handling such objects, you may pass your own reviver function to JSON.parse, for example to handle Date objects the following reviver can be used:

1import { RedisPubSub } from 'graphql-redis-subscriptions';
2
3const dateReviver = (key, value) => {
4  const isISO8601Z = /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}(?:\.\d*)?)Z$/;
5  if (typeof value === 'string' && isISO8601Z.test(value)) {
6    const tempDateNumber = Date.parse(value);
7    if (!isNaN(tempDateNumber)) {
8      return new Date(tempDateNumber);
9    }
10  }
11  return value;
12};
13
14const pubSub = new RedisPubSub({ ..., reviver: dateReviver });
15
16pubSub.publish('Test', {
17  validTime: new Date(),
18  invalidTime: '2018-13-01T12:00:00Z'
19});
20pubSub.subscribe('Test', message => {
21  message.validTime; // Javascript Date
22  message.invalidTime; // string
23});

Old Usage (Deprecated)

1import { RedisPubSub } from 'graphql-redis-subscriptions';
2const pubsub = new RedisPubSub();
3const subscriptionManager = new SubscriptionManager({
4  schema,
5  pubsub,
6  setupFunctions: {},
7});

Using Trigger Transform (Deprecated)

Recently, graphql-subscriptions package added a way to pass in options to each call of subscribe. Those options are constructed via the setupFunctions object you provide the Subscription Manager constructor. The reason for graphql-subscriptions to add that feature is to allow pub sub engines a way to reduce their subscription set using the best method of said engine. For example, Meteor's live query could use Mongo selector with arguments passed from the subscription like the subscribed entity id. For Redis, this could be a bit more simplified, but much more generic. The standard for Redis subscriptions is to use dot notations to make the subscription more specific. This is only the standard but I would like to present an example of creating a specific subscription using the channel options feature.

First I create a simple and generic trigger transform

1const triggerTransform = (trigger, {path}) => [trigger, ...path].join('.');

Then I pass it to the RedisPubSub constructor.

1const pubsub = new RedisPubSub({
2  triggerTransform,
3});

Lastly, I provide a setupFunction for commentsAdded subscription field. It specifies one trigger called comments.added and it is called with the channelOptions object that holds repoName path fragment.

1const subscriptionManager = new SubscriptionManager({
2  schema,
3  setupFunctions: {
4    commentsAdded: (options, {repoName}) => ({
5      'comments.added': {
6        channelOptions: {path: [repoName]},
7      },
8    }),
9  },
10  pubsub,
11});

When I call subscribe like this:

1const query = `
2  subscription X($repoName: String!) {
3    commentsAdded(repoName: $repoName)
4  }
5`;
6const variables = {repoName: 'graphql-redis-subscriptions'};
7subscriptionManager.subscribe({query, operationName: 'X', variables, callback});

The subscription string that Redis will receive will be comments.added.graphql-redis-subscriptions. This subscription string is much more specific and means the the filtering required for this type of subscription is not needed anymore. This is one step towards lifting the load off of the GraphQL API server regarding subscriptions.

Tests

Spin a Redis in docker server and cluster

Please refer to https://github.com/Grokzen/docker-redis-cluster documentation to start a cluster

1$ docker run --rm -p 6379:6379 redis:alpine
2$ export REDIS_CLUSTER_IP=0.0.0.0; docker run -e "IP=0.0.0.0" --rm -p 7006:7000 -p 7001:7001 -p 7002:7002 -p 7003:7003 -p 7004:7004 -p 7005:7005 grokzen/redis-cluster

Test

1npm run test