Subscriptions in Apollo Server

Subscriptions are long-lasting GraphQL read operations that can update their result whenever a particular server-side event occurs. Most commonly, updated results are pushed from the server to subscribing clients. For example, a chat application's server might use a subscription to push newly received messages to all clients in a particular chat room.

Because subscription updates are usually pushed by the server (instead of polled by the client), they generally use the WebSocket protocol instead of HTTP.

Schema definition

Your schema's Subscription type defines top-level fields that clients can subscribe to:

1
type Subscription {
2
  postCreated: Post
3
}

The postCreated field will update its value whenever a new Post is created on the backend, thus pushing the Post to subscribing clients.

Clients can subscribe to the postCreated field with a GraphQL string, like this:

1
subscription PostFeed {
2
  postCreated {
3
    author
4
    comment
5
  }
6
}

Enabling subscriptions

To run both an Express app and a separate WebSocket server for subscriptions, we'll create an http.Server instance that effectively wraps the two and becomes our new listener.

1. Install graphql-ws, ws, and @graphql-tools/schema:

   Bash

   1
   npm install graphql-ws ws @graphql-tools/schema

   Copy

2. Add the following imports to the file where you initialize your ApolloServer instance (we'll use these in later steps):

   TypeScript

   index.ts

   1
   import { createServer } from 'http';
   2
   import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer';
   3
   import { makeExecutableSchema } from '@graphql-tools/schema';
   4
   import { WebSocketServer } from 'ws';
   5
   import { useServer } from 'graphql-ws/lib/use/ws';

   Copy

3. Next, in order to set up both the HTTP and subscription servers, we need to first create an http.Server. Do this by passing your Express app to the createServer function, which we imported from the http module:

   TypeScript

   index.ts

   1
   // This `app` is the returned value from `express()`.
   2
   const httpServer = createServer(app);

   Copy

4. Create an instance of GraphQLSchema (if you haven't already).

   If you already pass the schema option to the ApolloServer constructor (instead of typeDefs and resolvers), you can skip this step.

   The subscription server (which we'll instantiate next) doesn't take typeDefs and resolvers options. Instead, it takes an executable GraphQLSchema. We can pass this schema object to both the subscription server and ApolloServer. This way, we make sure that the same schema is being used in both places.

   TypeScript

   index.ts

   1
   const schema = makeExecutableSchema({ typeDefs, resolvers });
   2
   // ...
   3
   const server = new ApolloServer({
   4
     schema,
   5
   });

   Copy

5. Create a WebSocketServer to use as your subscription server.

   TypeScript

   index.ts

   1
   // Creating the WebSocket server
   2
   const wsServer = new WebSocketServer({
   3
     // This is the `httpServer` we created in a previous step.
   4
     server: httpServer,
   5
     // Pass a different path here if app.use
   6
     // serves expressMiddleware at a different path
   7
     path: '/graphql',
   8
   });
   9
   
   10
   // Hand in the schema we just created and have the
   11
   // WebSocketServer start listening.
   12
   const serverCleanup = useServer({ schema }, wsServer);

   Copy

6. Add plugins to your ApolloServer constructor to shutdown both the HTTP server and the WebSocketServer:

   TypeScript

   index.ts

   1
   const server = new ApolloServer({
   2
     schema,
   3
     plugins: [
   4
       // Proper shutdown for the HTTP server.
   5
       ApolloServerPluginDrainHttpServer({ httpServer }),
   6
   
   7
       // Proper shutdown for the WebSocket server.
   8
       {
   9
         async serverWillStart() {
   10
           return {
   11
             async drainServer() {
   12
               await serverCleanup.dispose();
   13
             },
   14
           };
   15
         },
   16
       },
   17
     ],
   18
   });

   Copy

7. Finally, ensure you are listening to your httpServer.

   Most Express applications call app.listen(...), but for your setup change this to httpServer.listen(...) using the same arguments. This way, the server starts listening on the HTTP and WebSocket transports simultaneously.

A completed example of setting up subscriptions is shown below:

1
import { ApolloServer } from '@apollo/server';
2
import { expressMiddleware } from '@apollo/server/express4';
3
import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer';
4
import { createServer } from 'http';
5
import express from 'express';
6
import { makeExecutableSchema } from '@graphql-tools/schema';
7
import { WebSocketServer } from 'ws';
8
import { useServer } from 'graphql-ws/lib/use/ws';
9
import bodyParser from 'body-parser';
10
import cors from 'cors';
11
import resolvers from './resolvers';
12
import typeDefs from './typeDefs';
13

14
// Create the schema, which will be used separately by ApolloServer and
15
// the WebSocket server.
16
const schema = makeExecutableSchema({ typeDefs, resolvers });
17

18
// Create an Express app and HTTP server; we will attach both the WebSocket
19
// server and the ApolloServer to this HTTP server.
20
const app = express();
21
const httpServer = createServer(app);
22

23
// Create our WebSocket server using the HTTP server we just set up.
24
const wsServer = new WebSocketServer({
25
  server: httpServer,
26
  path: '/graphql',
27
});
28
// Save the returned server's info so we can shutdown this server later
29
const serverCleanup = useServer({ schema }, wsServer);
30

31
// Set up ApolloServer.
32
const server = new ApolloServer({
33
  schema,
34
  plugins: [
35
    // Proper shutdown for the HTTP server.
36
    ApolloServerPluginDrainHttpServer({ httpServer }),
37

38
    // Proper shutdown for the WebSocket server.
39
    {
40
      async serverWillStart() {
41
        return {
42
          async drainServer() {
43
            await serverCleanup.dispose();
44
          },
45
        };
46
      },
47
    },
48
  ],
49
});
50

51
await server.start();
52
app.use('/graphql', cors<cors.CorsRequest>(), bodyParser.json(), expressMiddleware(server));
53

54
const PORT = 4000;
55
// Now that our HTTP server is fully set up, we can listen to it.
56
httpServer.listen(PORT, () => {
57
  console.log(`Server is now running on http://localhost:${PORT}/graphql`);
58
});

1
import { ApolloServer } from '@apollo/server';
2
import { expressMiddleware } from '@apollo/server/express4';
3
import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer';
4
import { createServer } from 'http';
5
import express from 'express';
6
import { makeExecutableSchema } from '@graphql-tools/schema';
7
import { WebSocketServer } from 'ws';
8
import { useServer } from 'graphql-ws/lib/use/ws';
9
import bodyParser from 'body-parser';
10
import cors from 'cors';
11
import resolvers from './resolvers';
12
import typeDefs from './typeDefs';
13

14
// Create the schema, which will be used separately by ApolloServer and
15
// the WebSocket server.
16
const schema = makeExecutableSchema({ typeDefs, resolvers });
17

18
// Create an Express app and HTTP server; we will attach both the WebSocket
19
// server and the ApolloServer to this HTTP server.
20
const app = express();
21
const httpServer = createServer(app);
22

23
// Create our WebSocket server using the HTTP server we just set up.
24
const wsServer = new WebSocketServer({
25
  server: httpServer,
26
  path: '/graphql',
27
});
28
// Save the returned server's info so we can shutdown this server later
29
const serverCleanup = useServer({ schema }, wsServer);
30

31
// Set up ApolloServer.
32
const server = new ApolloServer({
33
  schema,
34
  plugins: [
35
    // Proper shutdown for the HTTP server.
36
    ApolloServerPluginDrainHttpServer({ httpServer }),
37

38
    // Proper shutdown for the WebSocket server.
39
    {
40
      async serverWillStart() {
41
        return {
42
          async drainServer() {
43
            await serverCleanup.dispose();
44
          },
45
        };
46
      },
47
    },
48
  ],
49
});
50

51
await server.start();
52
app.use('/graphql', cors(), bodyParser.json(), expressMiddleware(server));
53

54
const PORT = 4000;
55
// Now that our HTTP server is fully set up, we can listen to it.
56
httpServer.listen(PORT, () => {
57
  console.log(`Server is now running on http://localhost:${PORT}/graphql`);
58
});

Resolving a subscription

Resolvers for Subscription fields differ from resolvers for fields of other types. Specifically, Subscription field resolvers are objects that define a subscribe function:

1
const resolvers = {
2
  Subscription: {
3
    hello: {
4
      // Example using an async generator
5
      subscribe: async function* () {
6
        for await (const word of ['Hello', 'Bonjour', 'Ciao']) {
7
          yield { hello: word };
8
        }
9
      },
10
    },
11
    postCreated: {
12
      // More on pubsub below
13
      subscribe: () => pubsub.asyncIterator(['POST_CREATED']),
14
    },
15
  },
16
  // ...other resolvers...
17
};

The subscribe function must return an object of type AsyncIterator, a standard interface for iterating over asynchronous results. In the above postCreated.subscribe field, an AsyncIterator is generated by pubsub.asyncIterator (more on this below).

The PubSub class

You can use the publish-subscribe (pub/sub) model to track events that update active subscriptions. The graphql-subscriptions library provides the PubSub class as a basic in-memory event bus to help you get started:

To use the graphql-subscriptions package, first install it like so:

1
npm install graphql-subscriptions

A PubSub instance enables your server code to both publish events to a particular label and listen for events associated with a particular label. We can create a PubSub instance like so:

1
import { PubSub } from 'graphql-subscriptions';
2

3
const pubsub = new PubSub();

Publishing an event

You can publish an event using the publish method of a PubSub instance:

1
pubsub.publish('POST_CREATED', {
2
  postCreated: {
3
    author: 'Ali Baba',
4
    comment: 'Open sesame',
5
  },
6
});

• The first parameter is the name of the event label you're publishing to, as a string.
  • You don't need to register a label name before publishing to it.
• The second parameter is the payload associated with the event.
  • The payload should include whatever data is necessary for your resolvers to populate the associated Subscription field and its subfields.

When working with GraphQL subscriptions, you publish an event whenever a subscription's return value should be updated. One common cause of such an update is a mutation, but any back-end logic might result in changes that should be published.

As an example, let's say our GraphQL API supports a createPost mutation:

1
type Mutation {
2
  createPost(author: String, comment: String): Post
3
}

A basic resolver for createPost might look like this:

1
const resolvers = {
2
  Mutation: {
3
    createPost(parent, args, { postController }) {
4
      // Datastore logic lives in postController
5
      return postController.createPost(args);
6
    },
7
  },
8
  // ...other resolvers...
9
};

Before we persist the new post's details in our datastore, we can publish an event that also includes those details:

1
const resolvers = {
2
  Mutation: {
3
    createPost(parent, args, { postController }) {
4
      pubsub.publish('POST_CREATED', { postCreated: args }); 
5
      return postController.createPost(args);
6
    },
7
  },
8
  // ...other resolvers...
9
};

Next, we can listen for this event in our Subscription field's resolver.

Listening for events

An AsyncIterator object listens for events that are associated with a particular label (or set of labels) and adds them to a queue for processing.

You can create an AsyncIterator by calling the asyncIterator method of PubSub and passing in an array containing the names of the event labels that this AsyncIterator should listen for.

1
pubsub.asyncIterator(['POST_CREATED']);

Every Subscription field resolver's subscribe function must return an AsyncIterator object.

This brings us back to the code sample at the top of Resolving a subscription:

1
const resolvers = {
2
  Subscription: {
3
    postCreated: {
4
      subscribe: () => pubsub.asyncIterator(['POST_CREATED']),
5
    },
6
  },
7
  // ...other resolvers...
8
};

With this subscribe function set, Apollo Server uses the payloads of POST_CREATED events to push updated values for the postCreated field.

Filtering events

Sometimes, a client should only receive updated subscription data if that data meets certain criteria. To support this, you can call the withFilter helper function in your Subscription field's resolver.

Example

Let's say our server provides a commentAdded subscription, which should notify clients whenever a comment is added to a specified code repository. A client can execute a subscription that looks like this:

1
subscription ($repoName: String!) {
2
  commentAdded(repoFullName: $repoName) {
3
    id
4
    content
5
  }
6
}

This presents a potential issue: our server probably publishes a COMMENT_ADDED event whenever a comment is added to any repository. This means that the commentAdded resolver executes for every new comment, regardless of which repository it's added to. As a result, subscribing clients might receive data they don't want (or shouldn't even have access to).

To fix this, we can use the withFilter helper function to control updates on a per-client basis.

Here's an example resolver for commentAdded that uses the withFilter function:

1
import { withFilter } from 'graphql-subscriptions';
2

3
const resolvers = {
4
  Subscription: {
5
    commentAdded: {
6
      subscribe: withFilter(
7
        () => pubsub.asyncIterator('COMMENT_ADDED'),
8
        (payload, variables) => {
9
          // Only push an update if the comment is on
10
          // the correct repository for this operation
11
          return (
12
            payload.commentAdded.repository_name === variables.repoFullName
13
          );
14
        },
15
      ),
16
    },
17
  },
18
  // ...other resolvers...
19
};

The withFilter function takes two parameters:

• The first parameter is exactly the function you would use for subscribe if you weren't applying a filter.
• The second parameter is a filter function that returns true if a subscription update should be sent to a particular client, and false otherwise (Promise<boolean> is also allowed). This function takes two parameters of its own:
  • payload is the payload of the event that was published.
  • variables is an object containing all arguments the client provided when initiating their subscription.

Use withFilter to make sure clients get exactly the subscription updates they want (and are allowed to receive).

Basic runnable example

An example server is available on GitHub and CodeSandbox:

The server exposes one subscription (numberIncremented) that returns an integer that's incremented on the server every second. Here's an example subscription that you can run against your server:

1
subscription IncrementingNumber {
2
  numberIncremented
3
}

After you start up the server in CodeSandbox, follow the instructions in the browser to test running the numberIncremented subscription in Apollo Sandbox. You should see the subscription's value update every second.

Operation context

When initializing context for a query or mutation, you usually extract HTTP headers and other request metadata from the req object provided to the context function.

For subscriptions, you can extract information from a client's request by adding options to the first argument passed to the useServer function.

For example, you can provide a context property to add values to your GraphQL operation contextValue:

1
// ...
2
useServer(
3
  {
4
    // Our GraphQL schema.
5
    schema,
6
    // Adding a context property lets you add data to your GraphQL operation contextValue
7
    context: async (ctx, msg, args) => {
8
      // You can define your own function for setting a dynamic context
9
      // or provide a static value
10
      return getDynamicContext(ctx, msg, args);
11
    },
12
  },
13
  wsServer,
14
);

Notice that the first parameter passed to the context function above is ctx. The ctx object represents the context of your subscription server, not the GraphQL operation contextValue that's passed to your resolvers.

You can access the parameters of a client's subscription request to your WebSocket server via the ctx.connectionParams property.

Below is an example of the common use case of extracting an authentication token from a client subscription request and using it to find the current user:

1
const getDynamicContext = async (ctx, msg, args) => {
2
  // ctx is the graphql-ws Context where connectionParams live
3
  if (ctx.connectionParams.authentication) {
4
    const currentUser = await findUser(ctx.connectionParams.authentication);
5
    return { currentUser };
6
  }
7
  // Otherwise let our resolvers know we don't have a current user
8
  return { currentUser: null };
9
};
10

11
useServer(
12
  {
13
    schema,
14
    context: async (ctx, msg, args) => {
15
      // Returning an object will add that information to
16
      // contextValue, which all of our resolvers have access to.
17
      return getDynamicContext(ctx, msg, args);
18
    },
19
  },
20
  wsServer,
21
);

Putting it all together, the useServer.context function returns an object, contextValue, which is available to your resolvers.

Note that the context option is called once per subscription request, not once per event emission. This means that in the above example, every time a client sends a subscription operation, we check their authentication token.

onConnect and onDisconnect

You can configure the subscription server's behavior whenever a client connects (onConnect) or disconnects (onDisconnect).

Defining an onConnect function enables you to reject a particular incoming connection by returning false or by throwing an exception. This can be helpful if you want to check authentication when a client first connects to your subscription server.

You provide these functions as options to the first argument of useServer, like so:

1
useServer(
2
  {
3
    schema,
4
    // As before, ctx is the graphql-ws Context where connectionParams live.
5
    onConnect: async (ctx) => {
6
      // Check authentication every time a client connects.
7
      if (tokenIsNotValid(ctx.connectionParams)) {
8
        // You can return false to close the connection  or throw an explicit error
9
        throw new Error('Auth token missing!');
10
      }
11
    },
12
    onDisconnect(ctx, code, reason) {
13
      console.log('Disconnected!');
14
    },
15
  },
16
  wsServer,
17
);

For more examples of using onConnect and onDisconnect, see the graphql-ws recipes documentation.

Example: Authentication with Apollo Client

In Apollo Client, the GraphQLWsLink constructor supports adding information to connectionParams (example). Those connectionParams are sent to your server when it connects, allowing you to extract information from the client request by accessing Context.connectionParams.

Let's suppose we create our subscription client like so:

1
import { GraphQLWsLink } from '@apollo/client/link/subscriptions';
2
import { createClient } from 'graphql-ws';
3

4
const wsLink = new GraphQLWsLink(
5
  createClient({
6
    url: 'ws://localhost:4000/subscriptions',
7
    connectionParams: {
8
      authentication: user.authToken,
9
    },
10
  }),
11
);

The connectionParams argument (which contains the information provided by the client) is passed to your server, enabling you to validate the user's credentials.

From there you can use the useServer.context property to authenticate the user, returning an object that's passed into your resolvers as the context argument during execution.

For our example, we can use the connectionParams.authentication value provided by the client to look up the related user before passing that user along to our resolvers:

1
const findUser = async (authToken) => {
2
  // Find a user by their auth token
3
};
4

5
const getDynamicContext = async (ctx, msg, args) => {
6
  if (ctx.connectionParams.authentication) {
7
    const currentUser = await findUser(ctx.connectionParams.authentication);
8
    return { currentUser };
9
  }
10
  // Let the resolvers know we don't have a current user so they can
11
  // throw the appropriate error
12
  return { currentUser: null };
13
};
14

15
// ...
16
useServer(
17
  {
18
    // Our GraphQL schema.
19
    schema,
20
    context: async (ctx, msg, args) => {
21
      // This will be run every time the client sends a subscription request
22
      return getDynamicContext(ctx, msg, args);
23
    },
24
  },
25
  wsServer,
26
);

To sum up, the example above looks up a user based on the authentication token sent with each subscription request before returning the user object to be used by our resolvers. If no user exists or the lookup otherwise fails, our resolvers can throw an error and the corresponding subscription operation is not executed.

Production PubSub libraries

As mentioned above, the PubSub class is not recommended for production environments, because its event-publishing system is in-memory. This means that events published by one instance of your GraphQL server are not received by subscriptions that are handled by other instances.

Instead, you should use a subclass of the PubSubEngine abstract class that you can back with an external datastore such as Redis or Kafka.

The following are community-created PubSub libraries for popular event-publishing systems:

• Redis
• Google PubSub
• MQTT enabled broker
• RabbitMQ
• Kafka
• Postgres
• Google Cloud Firestore
• Ably Realtime
• Google Firebase Realtime Database
• Azure SignalR Service
• Azure ServiceBus
• MongoDB

If none of these libraries fits your use case, you can also create your own PubSubEngine subclass. If you create a new open-source library, click Edit on GitHub to let us know about it!

Switching from subscriptions-transport-ws

This article previously demonstrated using the subscriptions-transport-ws library to set up subscriptions. However, this library is no longer actively maintained. You can still use it with Apollo Server, but we strongly recommend using graphql-ws instead.

For details on how to switch from subscriptions-transport-ws to graphql-ws, follow the steps in the Apollo Server 3 docs.

Updating subscription clients

If you intend to switch from subscriptions-transport-ws to graphql-ws you will need to update the following clients: