Updated graphql event stream functionality to watch for changes to

schema

Author: imolorhe

.gitignore

@@ -1,6 +1,9 @@
 build
+dumps
 .vscode
 
+.DS_Store
+
 # Logs
 logs
 *.log

package.json

@@ -7,22 +7,26 @@
   "license": "MIT",
   "scripts": {
     "build": "tsc",
-    "dev": "nodemon src/index.ts"
+    "dev": "nodemon src/index.ts",
+    "dev:rs": "nodemon src/rs-index.ts --ignore src/schema/"
   },
   "dependencies": {
-    "apollo-server-express": "^2.14.2",
+    "apollo-server-express": "^2.19.1",
     "axios": "^0.18.1",
+    "chokidar": "^3.5.0",
     "compression": "^1.7.4",
+    "cors": "^2.8.5",
     "express": "^4.17.1",
     "graphql": "^14.2.0",
+    "graphql-tools": "^7.0.2",
     "lodash": "^4.17.20"
   },
   "devDependencies": {
     "@types/compression": "^1.7.0",
     "@types/express-serve-static-core": "^4.17.7",
     "@types/graphql": "^14.2.0",
     "@types/lodash": "^4.14.123",
-    "nodemon": "^1.18.10",
+    "nodemon": "^2.0.7",
     "ts-node": "^8.0.3",
     "typescript": "3.4.1"
   },

src/index.ts

@@ -2,8 +2,8 @@ import http from 'http';
 import express, { RequestHandler } from 'express';
 import { ApolloServer, ApolloError } from 'apollo-server-express';
 import { typeDefs, resolvers } from './schema';
-import { graphqlEventStream } from './schema-observer';
 const compression = require('compression');
+const cors = require('cors');
 
 const server = new ApolloServer({
   typeDefs,
@@ -23,17 +23,19 @@ const server = new ApolloServer({
 });
 const app = express();
 app.use(compression());
+app.use(cors());
 app.use((req, res, next) => {
     // console.log(req.headers);
     return next();
 });
-app.use(graphqlEventStream());
+// app.use(graphqlEventStream());
+
 server.applyMiddleware({ app, path: '/graphql' });
 const httpServer = http.createServer(app);
 server.installSubscriptionHandlers(httpServer);
 app.all('/test', (req, res) => {
     // console.log(req.headers);
-    res.send('Hi');
+    res.send('');
 });
 const PORT = process.env.__PORT__ || 5400;
 httpServer.listen(PORT, () => {

src/rs-index.ts

@@ -0,0 +1,99 @@
+import http from 'http';
+import { EventEmitter } from 'events';
+import express, { RequestHandler, Express } from 'express';
+import { ApolloServer, ApolloError } from 'apollo-server-express';
+import { graphqlEventStream } from './schema-observer';
+import { execute } from 'graphql';
+import { makeExecutableSchema } from 'graphql-tools';
+import { createWatcher } from './watcher';
+
+const compression = require('compression');
+const cors = require('cors');
+
+const getSchema = () => {
+  // First clear require cache for the module
+  delete require.cache[require.resolve('./schema/rs-schema')];
+  const { typeDefs, resolvers } = require('./schema/rs-schema');
+  return makeExecutableSchema({
+    typeDefs,
+    resolvers,
+  })
+};
+
+const setupServer = async() => {
+  const watcher = createWatcher();
+  const emitter = new EventEmitter();
+
+  const server = new ApolloServer({
+    subscriptions: false,
+    gateway: {
+      load: async() => {
+        const schema = getSchema();
+        return {
+          schema,
+          executor: args => {
+            return execute({
+              ...args,
+              schema,
+            })
+          }
+        }
+      },
+      executor: execute,
+      onSchemaChange: cb => {
+        watcher.on('all', () => {
+          console.log('updating schema..');
+          const schema = getSchema();
+          cb(schema);
+          emitter.emit('schema:updated', schema);
+        });
+
+        return () => watcher.close();
+      },
+    },
+    context({ req }) {
+      // throw new ApolloError('NOT_FOUND', '404');
+      return { req };
+    },
+    formatError(error) {
+      if (!process.env.NODE_ENV || process.env.NODE_ENV === 'development') {
+        // logging the errors can help in development
+        // tslint:disable-next-line
+        console.log(error);
+      }
+      return error;
+    }
+  });
+  const app = express();
+  
+  // Compression doesn't work well with server side events
+  // app.use(compression());
+  app.use(cors());
+  app.use((req, res, next) => {
+      // console.log(req.headers);
+      return next();
+  });
+  graphqlEventStream({
+    app,
+    emitter,
+    streamPath: '/stream',
+  });
+  server.applyMiddleware({ app, path: '/graphql' });
+  const httpServer = http.createServer(app);
+  // subscriptions are not supported in apollo gateway
+  // server.installSubscriptionHandlers(httpServer);
+  
+  app.all('/test', (req, res) => {
+      // console.log(req.headers);
+      res.send('');
+  });
+  const PORT = process.env.__PORT__ || 5400;
+  httpServer.listen(PORT, () => {
+    // tslint:disable-next-line
+    console.log(`Server running at http://localhost:${PORT}${server.graphqlPath}`);
+    // tslint:disable-next-line
+    // console.log(`Subscription running at ws://localhost:${PORT}${server.subscriptionsPath}`);
+  });
+};
+
+setupServer();

src/schema-observer.ts

@@ -1,49 +1,45 @@
-import { Request, Response } from 'express';
+import { Express } from 'express';
+import { EventEmitter } from 'events';
 
-export const graphqlEventStream = ({ streamPath = '/stream' } = {}) => {
-  return (req: Request, res: Response) => {
-    const sse = (str = '') => {
-      res.write(str);
-      res.flushHeaders();
-    };
-    const cleanup = () => {
-      // tslint:disable-next-line
-      console.log('Cleaning up...');
-      res.end();
-    };
-
-    // If current request is for the stream, setup the stream.
-    if (req.path === streamPath) {
-      if (req.headers.accept !== 'text/event-stream') {
-        res.statusCode = 405;
-        res.end();
-        return;
-      }
-
-      // tslint:disable-next-line
-      console.log('stream requested.');
-      // Making sure these options are set.
-      req.socket.setTimeout(0);
-      req.socket.setNoDelay(true);
-      req.socket.setKeepAlive(true);
-
-      // Set headers for Server-Sent Events.
-      res.statusCode = 200;
-      res.setHeader('Content-Type', 'text/event-stream');
-      res.setHeader('Cache-Control', 'no-cache');
-      res.setHeader('Connection', 'keep-alive');
-
-      sse('event: open\n\n');
+export const graphqlEventStream = ({
+  streamPath = '/stream',
+  emitter,
+  app,
+}: { app: Express, streamPath: string, emitter: EventEmitter }) => {
+  app.use((req, res) => {
+    res.setHeader('X-GraphQL-Event-Stream', streamPath);
+    if (req.next) {
+      req.next();
+    }
+  });
 
-      req.on('close', cleanup);
-      req.on('finish', cleanup);
-      req.on('error', cleanup);
+  app.get(streamPath, (req, res) => {
+    if (req.headers.accept !== 'text/event-stream') {
+      res.statusCode = 405;
+      res.end();
       return;
     }
 
-    res.setHeader('X-GraphQL-Event-Stream', streamPath);
-    if (req.next) {
-      return req.next();
-    }
-  };
+    res.setHeader('Cache-Control', 'no-cache');
+    res.setHeader('Content-Type', 'text/event-stream');
+    res.setHeader('Connection', 'keep-alive');
+    res.setHeader('Access-Control-Allow-Origin', '*');
+    res.flushHeaders(); // flush the headers to establish SSE with client
+  
+    console.log('connected to client.');
+    let counter = 0;
+    const updateClients = () => {
+      counter++;
+      console.log('sending update to client..');
+      res.write(`data: ${JSON.stringify({num: counter})}\n\n`); // res.write() instead of res.send()
+    };
+    emitter.on('schema:updated', updateClients);
+  
+    // If client closes connection, stop sending events
+    res.on('close', () => {
+        console.log('client dropped me.');
+        emitter.off('schema:updated', updateClients);
+        res.end();
+    });
+  });
 };

src/schema/rs-schema.ts

@@ -0,0 +1,44 @@
+import { gql } from 'apollo-server-express';
+import { merge } from 'lodash';
+
+import * as GOTCharacter from './GOTCharacter';
+import * as GOTBook from './GOTBook';
+import * as GOTHouse from './GOTHouse';
+import * as HNUser from './HNUser';
+import * as Message from './Message';
+
+export const rootTypeDef = gql`
+  type Query {
+    hello: String
+    bye: Boolean
+  }
+  type Mutation
+  type Subscription
+`;
+export const rootResolvers = {
+  Query: {
+    hello: () => 'Hello world',
+    bye: () => true,
+  }
+};
+
+export const typeDefs = [
+  rootTypeDef,
+  GOTCharacter.typeDef,
+  GOTBook.typeDef,
+  GOTHouse.typeDef,
+  HNUser.typeDef,
+  Message.typeDef,
+  // for some reason file Upload type is causing problems
+  // File.typeDef,
+];
+
+export const resolvers = merge(
+  rootResolvers,
+  GOTCharacter.resolvers,
+  GOTBook.resolvers,
+  GOTHouse.resolvers,
+  HNUser.resolvers,
+  Message.resolvers,
+  // File.resolvers,
+);

src/watcher.ts

@@ -0,0 +1,6 @@
+import path from 'path';
+import chokidar from 'chokidar';
+
+export const createWatcher = () => {
+  return chokidar.watch(path.resolve(__dirname, './schema'));
+};