Subscriptions | GraphQL Modules

Subscription is a third kind of GraphQL operation, next to Query and Mutation. Think of it as an event emitter where events are emitted by your backend and consumed by a GraphQL client-frontend in most scenarios.

GraphQL is just a specification and GraphQL Yoga or Apollo Server are Node implementations of the server-side part of GraphQL. A natural fit for Subscriptions is the WebSocket protocol, but it could be anything. We’re going to focus only on WS and use subscriptions-transport-ws package as our transport layer.

Setup

For Queries and Mutations, we use createExecution() method to create an execution logic, in the case of Subscriptions it’s very similar. We construct the subscription phase with createSubscription() method and provide it to a GraphQL server.

import { createApplication } from 'graphql-modules'

const application = createApplication({
  modules: [
    // ...
  ]
})

const subscribe = application.createSubscription()

Why do we need it? GraphQL Modules need to understand the life cycle of a subscription to avoid memory leaks.

Example

We recommend using graphql-subscriptions as it provides a handful utility functions (filtering for example) and an event emitter.

Provide PubSub with an instance as the value so as a singleton service it becomes available across all modules.

application.ts

import { createApplication } from 'graphql-modules'
import { PubSub } from 'graphql-subscriptions'
import { myModule } from './my-module'

const application = createApplication({
  modules: [myModule /* ... */],
  providers: [
    {
      provide: PubSub,
      useValue: new PubSub()
    }
  ]
})

Accessing PubSub in a module can be done both, in a resolve function and Messages service.

Take a look at two things here, how MESSAGE_ADDED event was emitter in Messages.send() and how Subscriptions.messageAdded subscribes to events.

my-module.ts

import { createModule, Injectable, gql } from 'graphql-modules'
import { PubSub } from 'graphql-subscriptions'

const MESSAGE_ADDED = 'MESSAGE_ADDED'

const messages = [
  // ...
]

@Injectable()
class Messages {
  constructor(private pubsub: PubSub) {}

  async all() {
    return messages
  }

  async send(body: string) {
    const message = {
      id: generateRandomId(),
      body
    }

    messages.push(message)

    this.pubsub.publish(MESSAGE_ADDED, { messageAdded: message })

    return message
  }
}

export const myModule = createModule({
  id: 'my-module',
  providers: [Messages],
  typeDefs: gql`
    type Query {
      messages: [Message!]
    }

    type Mutation {
      sendMessage(message: String!): Message!
    }

    type Subscription {
      messageAdded: Message
    }

    type Message {
      id: ID!
      body: String!
    }
  `,
  resolvers: {
    Query: {
      messages(parent, args, ctx: GraphQLModules.Context) {
        return ctx.injector.get(Messages).all()
      }
    },
    Mutation: {
      sendMessage(parent, { message }, ctx: GraphQLModules.Context) {
        return ctx.injector.get(Messages).send(message)
      }
    },
    Subscription: {
      messageAdded: {
        subscribe(root, args, ctx: GraphQLModules.Context) {
          return ctx.injector.get(PubSub).asyncIterator([MESSAGE_ADDED])
        }
      }
    }
  }
})

Here are reference implementations of using GraphQL Subscriptions with WebSockets in both, Apollo Server and Express GraphQL.

server.ts

import 'reflect-metadata'
import { createServer } from '@graphql-yoga/node'
import { useGraphQLModules } from '@envelop/graphql-modules'
import { application } from './application'

const server = createServer({
  plugins: [useGraphQLModules(application)]
})

server.start().then(() => {
  console.log(`🚀 Server ready`)
})

server.ts

import 'reflect-metadata'
import express from 'express'
import { createServer } from 'node:http'
import { SubscriptionServer } from 'subscriptions-transport-ws'
import { ApolloServer } from 'apollo-server-express'
import { application } from './application'

const schema = application.createSchemaForApollo()

const { schema, createExecution, createSubscription, createApolloExecutor } =
  application
const execute = createExecution()
const subscribe = createSubscription()

;(async function () {
  const app = express()

  const httpServer = createServer(app)

  let subscriptionServer
  const server = new ApolloServer({
    schema,
    executor: createApolloExecutor(),
    plugins: [
      {
        serverWillStart() {
          return {
            drainServer() {
              subscriptionServer.close()
            }
          }
        }
      }
    ]
  })

  subscriptionServer = SubscriptionServer.create(
    {
      schema,
      execute,
      subscribe
    },
    {
      server: httpServer,
      path: server.graphqlPath
    }
  )

  await server.start()
  server.applyMiddleware({ app })

  const PORT = 4000
  httpServer.listen(PORT, () =>
    console.log(`Server is now running on http://localhost:${PORT}/graphql`)
  )
})()

server.ts

import 'reflect-metadata'
import express from 'express'
import graphqlHTTP from 'express-graphql'
import { createServer } from 'node:http'
import { SubscriptionServer } from 'subscriptions-transport-ws'
import { application } from './application'

const execute = application.createExecution()
const subscribe = application.createSubscription()
const { schema } = application

const server = express()

server.use(
  '/',
  graphqlHTTP({
    schema,
    customExecuteFn: execute,
    graphiql: true
  })
)

const webServer = createServer(app)

webServer.listen(4000, () => {
  console.log('🚀 Server ready at http://localhost:4000')

  new SubscriptionServer(
    {
      execute,
      subscribe,
      schema
    },
    {
      server: webServer,
      path: '/'
    }
  )
})