Subscriptions

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 usually use the WebSocket protocol instead of HTTP. To support this, Apollo Server lets you set a subscription-specific endpoint that's separate from the default endpoint for queries and mutations.

You can use subscriptions with the core apollo-server library, or with any of Apollo Server's supported middleware integrations.

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
}

Setting a subscription endpoint

Because subscriptions use WebSocket instead of HTTP, Apollo Server uses a second GraphQL endpoint specifically for subscriptions. This endpoint uses the ws protocol instead of http.

By default, the subscription endpoint's path matches the path of your primary GraphQL endpoint (/graphql if not set). You can specify a different path for your subscription endpoint like so:

1
const server = new ApolloServer({
2
  subscriptions: {
3
    path: '/subscriptions'
4
  },
5
  // ...other options...
6
}));

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
    postCreated: {
4
      // More on pubsub below
5
      subscribe: () => pubsub.asyncIterator(['POST_CREATED']),
6
    },
7
  },
8
  // ...other resolvers...
9
};

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

The PubSub class

Apollo Server uses a publish-subscribe (pub/sub) model to track events that update active subscriptions. The graphql-subscriptions library (included in every apollo-server package) provides the PubSub class as a basic in-memory event bus to help you get started:

1
const { PubSub } = require('apollo-server');
2

3
const pubsub = new PubSub();

A PubSub instance enables your server code to both publish events to a particular label and listen for events associated with a particular label.

Publishing an event

You publish an event with 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, context) {
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, context) {
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 create an AsyncIterator by calling the asyncIterator method of PubSub:

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

You pass this method an array containing the names of all event labels that the AsyncIterator should listen for.

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
const { withFilter } = require('apollo-server');
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 (payload.commentAdded.repository_name === variables.repoFullName);
12
        },
13
      ),
14
    }
15
  },
16
    // ...other resolvers...
17
};

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. The example requires only the apollo-server library.

After you start up this server locally, you can visit http://localhost:4000 to test out running the numberIncremented subscription in GraphQL Playground. You'll 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 extract this metadata from the connection object instead. This object adheres to the ExecutionParams interface.

Because all operation types use the same context initialization function, you should check which of req or connection is present for each incoming request:

1
const server = new ApolloServer({
2
  context: ({ req, connection }) => {
3
    if (connection) { // Operation is a Subscription
4
      // Obtain connectionParams-provided token from connection.context
5
      const token = connection.context.authorization || "";
6
      return { token };
7
    } else { // Operation is a Query/Mutation
8
      // Obtain header-provided token from req.headers
9
      const token = req.headers.authorization || "";
10
      return { token };
11
    }
12
  },
13
});

This is especially important because metadata like auth tokens are sent differently depending on the transport.

onConnect and onDisconnect

You can define functions that Apollo Server executes whenever a subscription request connects (onConnect) or disconnects (onDisconnect).

Defining an onConnect function provides the following benefits:

• You can reject a particular incoming connection by throwing an exception or returning false in onConnect.
  • This is especially useful for authentication.
• If onConnect returns an object, that object's fields are added to the WebSocket connection's context object.
  • This is not the operation context that's passed between resolvers. However, you can transfer these values from the connection's context when you initialize operation context.

You provide these function definitions to the constructor of ApolloServer, like so:

1
const server = new ApolloServer({
2
  subscriptions: {
3
    onConnect: (connectionParams, webSocket, context) => {
4
      console.log('Connected!')
5
    },
6
    onDisconnect: (webSocket, context) => {
7
      console.log('Disconnected!')
8
    },
9
    // ...other options...
10
  },
11
});

These functions are passed the following parameters:

Example: Authentication with onConnect

On the client, SubscriptionsClient supports adding token information to connectionParams (example) that will be sent with the first WebSocket message. In the server, all GraphQL subscriptions are delayed until the connection has been fully authenticated and the onConnect callback returns a truthy value.

The connectionParams argument in the onConnect callback contains the information passed by the client and can be used to validate user credentials. The GraphQL context can also be extended with the authenticated user data to enable fine grain authorization.

1
const { ApolloServer } = require('apollo-server');
2
const { resolvers, typeDefs } = require('./schema');
3

4
const validateToken = authToken => {
5
  // ... validate token and return a Promise, rejects in case of an error
6
};
7

8
const findUser = authToken => {
9
  return tokenValidationResult => {
10
    // ... finds user by auth token and return a Promise, rejects in case of an error
11
  };
12
};
13

14
const server = new ApolloServer({
15
  typeDefs,
16
  resolvers,
17
  subscriptions: {
18
    onConnect: (connectionParams, webSocket) => {
19
      if (connectionParams.authToken) {
20
        return validateToken(connectionParams.authToken)
21
          .then(findUser(connectionParams.authToken))
22
          .then(user => {
23
            return {
24
              currentUser: user,
25
            };
26
          });
27
      }
28

29
      throw new Error('Missing auth token!');
30
    },
31
  },
32
});
33

34
server.listen().then(({ url, subscriptionsUrl }) => {
35
  console.log(`🚀 Server ready at ${url}`);
36
  console.log(`🚀 Subscriptions ready at ${subscriptionsUrl}`);
37
});

The example above validates the user's token that is sent with the first initialization message on the transport, then it looks up the user and returns the user object as a Promise. The user object found will be available as context.currentUser in your GraphQL resolvers.

In case of an authentication error, the Promise will be rejected, which prevents the client's connection.

Using with middleware integrations

You can use subscriptions with any of Apollo Server's supported middleware integrations. To do so, you call installSubscriptionHandlers on your ApolloServer instance.

This example enables subscriptions for an Express server that uses apollo-server-express:

1
const http = require('http');
2
const { ApolloServer } = require('apollo-server-express');
3
const express = require('express');
4

5
async function startApolloServer() {
6
  const PORT = 4000;
7
  const app = express();
8
  const server = new ApolloServer({ typeDefs, resolvers });
9
  await server.start();
10
  server.applyMiddleware({app})
11

12
  const httpServer = http.createServer(app);
13
  server.installSubscriptionHandlers(httpServer); 
14

15
  // Make sure to call listen on httpServer, NOT on app.
16
  await new Promise(resolve => httpServer.listen(PORT, resolve));
17
  console.log(`🚀 Server ready at http://localhost:${PORT}${server.graphqlPath}`);
18
  console.log(`🚀 Subscriptions ready at ws://localhost:${PORT}${server.subscriptionsPath}`);
19
  return { server, app, httpServer };
20
}

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

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!