GraphQL Subscription with NodeJS and ReactJS apollo client

Hello Guys, this is another post for GraphQL Subscription with NodeJS and ReactJS apollo client. You might be wondering what new thing I am presenting here. So for reactjs apollo-client implementation and understanding please check here. For this demo, we have used  graphql-server-express, MySQL and React JS Apollo-Client.

• MySQL Database implementation.
• Complete CRUD operation with Subscription Support.
• Used all the latest npm packages till date. Below is what you will see after running example code attached in git.

Main Npm Packages used are as below

• MySQL : For connection to MySQL
• GraphQL-tools : To create executable schema which will be consumed by graphql-server-express
• GraphQL : Provides actual GraphQL query execute, subscribe
• graphql-server-express : graphqlExpress consumes schema and provides inbuilt graphql playground from graphqlExpress
• graphql-subscriptions : provides beautiful PubSub Method which makes our life much easier for subscription event firing.
• subscriptions-transport-ws : Actual socket server which uses graphql execute, subscribe and schema exposed by GraphQL and graphql-tools.

So let’s dive in to the code without wasting time. So first thing we need to do is create schema.graphql file and define it as below.

type Channel {
  id: Int!
  name: String!
}

type Query {
  channels: [Channel!]!
}

type Mutation {
  addChannel(name: String!): Channel!
  updateChannel(id: Int!, name: String!): Channel!
  deleteChannel(id: Int!): Channel!
}

type Subscription {
  subscriptionChannelAdded: Channel!
  subscriptionChannelUpdated: Channel!
  subscriptionChannelDeleted: Channel!
}

then create schema.js and resolver.js to read this schema.graphql file and creating graphql schema.

resolver.js

This file includes actual connection to database and queries for CRUD operation and firing subscription insert, update and delete. We will try to understand channelAdded Subscription.
You can see how subscriptionChannelAdded is defined in schema.graphql and in below
const CHANNEL_ADDED_TOPIC = ‘subscriptionChannelAdded’;

import { PubSub } from 'graphql-subscriptions';
var mysql = require('mysql')
var connection = mysql.createConnection({
  host: 'localhost',
  user: 'root',
  password: 'root',
  database: 'gqlgenreact'
});

const CHANNEL_ADDED_TOPIC = 'subscriptionChannelAdded';
const CHANNEL_UPDATE_TOPIC = 'subscriptionChannelUpdated';
const CHANNEL_DELETE_TOPIC = 'subscriptionChannelDeleted';
const pubsub = new PubSub();

export const resolvers = {
  Query: {
    channels: () => {
      return new Promise((resolve, reject) => {
        connection.query('SELECT * from channel', 
             function (err, rows, fields) {
          if (err) throw err
          resolve(rows)
        })
      });
    }
  },
  Mutation: {
    addChannel: (root, args) => {
      return new Promise((resolve, reject) => {
        connection.query('INSERT INTO channel (name) VALUES (?)', 
            [args.name], function (err, rows) {
          if (err) throw err
          const newChannel = { name: args.name, id: rows.insertId };
          console.log("newChannel :", newChannel);
          pubsub.publish(CHANNEL_ADDED_TOPIC, { 
            subscriptionChannelAdded: newChannel });
          resolve(newChannel)
        })
      });
    }, updateChannel: (root, args) => {
      return new Promise((resolve, reject) => {
        connection.query('UPDATE channel SET name=? WHERE id=?', 
             [args.name, args.id], function (err, rows) {
          if (err) throw err
          const newChannel = { name: args.name, id: args.id };
          pubsub.publish(CHANNEL_UPDATE_TOPIC, { 
             subscriptionChannelUpdated: newChannel });
          resolve(newChannel)
        })
      });

    }, deleteChannel: (root, args) => {
      return new Promise((resolve, reject) => {
        console.log(args);
        connection.query('DELETE FROM channel WHERE id=?', [args.id], 
            function (err, rows) {
          console.log(err);
          if (err) throw err
          const newChannel = { name: "", id: args.id };
          console.log(newChannel);
          pubsub.publish(CHANNEL_DELETE_TOPIC, { 
             subscriptionChannelDeleted: newChannel });
          resolve(newChannel)
        })
      });

    }
  },
  Subscription: {
    subscriptionChannelAdded: {
      subscribe: () => pubsub.asyncIterator(CHANNEL_ADDED_TOPIC)
    },
    subscriptionChannelUpdated: {
      subscribe: () => pubsub.asyncIterator(CHANNEL_UPDATE_TOPIC)
    },
    subscriptionChannelDeleted: {
      subscribe: () => pubsub.asyncIterator(CHANNEL_DELETE_TOPIC)
    }
  }
};

Once channelAdded Mutation is called we are firing pubsub.publish(CHANNEL_ADDED_TOPIC, { subscriptionChannelAdded: newChannel }); this will send event to all the subscribed sockets.

schema.js

This file connects schema.graphql and resolver.js file and creates  GraphQL executable schema which will be used by graphqlExpress.
we just need to notice two main line here, which read schema.graphql and converts it to typeDefs.
const schemaFile = path.join(__dirname, ‘../schema.graphql’);const typeDefs = fs.readFileSync(schemaFile, ‘utf8’);

const fs = require('fs');
const path = require('path');

import { makeExecutableSchema } from 'graphql-tools';

import { resolvers } from './resolvers';

const schemaFile = path.join(__dirname, '../schema.graphql');
const typeDefs = fs.readFileSync(schemaFile, 'utf8');

const schema = makeExecutableSchema({ typeDefs, resolvers });
export { schema };

server.js

Finally here is our main file which will start subscription server using subscriptions-transport-ws. This code is mostly common, you will be able to find it in graphql express official docs also.

import express from 'express';
import {
  graphqlExpress,
  graphiqlExpress,
} from 'graphql-server-express';
import bodyParser from 'body-parser';
import cors from 'cors';

import { schema } from './src/schema';

import { execute, subscribe } from 'graphql';
import { createServer } from 'http';
import { SubscriptionServer } from 'subscriptions-transport-ws';

const PORT = 8090;
const server = express();

server.use('*', cors({ origin: 'http://localhost:3000' }));

server.use('/graphql', bodyParser.json(), graphqlExpress({
  schema
}));

server.use('/graphiql', graphiqlExpress({
  endpointURL: '/graphql',
  subscriptionsEndpoint: `ws://localhost:${PORT}/query`
}));

// We wrap the express server so that we can attach the 
// WebSocket for subscriptions
const ws = createServer(server);

ws.listen(PORT, () => {
  console.log(`GraphQL Server is now running on 
  http://localhost:${PORT}`);

  // Set up the WebSocket for handling GraphQL subscriptions
  new SubscriptionServer({
    execute,
    subscribe,
    schema
  }, {
    server: ws,
    path: '/query',
  });
});

Recap for subscription part

#1  Defined below in schema.graphql
type Subscription { subscriptionChannelAdded: Channel!  }

#2    Use PubSub method provided by graphql-subscriptions which will help us to Subscribe and publish and will provide mechanism like EventEmitter.

import { PubSub } from 'graphql-subscriptions';
export const pubsub = new PubSub();

#3   Implement resolver for subscription type to map event using pubsub.asyncIterator. Once we request subscription from react apollo client, it will add our socket to its listening socket list and will send back events while we will call pubsub.publish

Subscription: {
    subscriptionChannelAdded: {
      subscribe: () => pubsub.asyncIterator(CHANNEL_ADDED_TOPIC)
    }
 }

#4 Call pubsub.publish method from Channel Added Mutation. This is tricky part don’t forgot to add it.
pubsub.publish(CHANNEL_ADDED_TOPIC, {subscriptionChannelAdded: newChannel});

Finally I have github code ready to check on with proper instruction on how to run it. Feel free to ask question or comment below.