feat!: add websocket server option

BREAKING CHANGE: the subscription server export is now an abstract class

README.md

@@ -1 +1,36 @@
 # GraphQL Subscription Server
+
+A subscription server for GraphQL subscriptions. Supports streaming over plain web sockets
+or Socket.IO, and integrates with Redis or any other Pub/Sub service.
+
+## Setup
+
+### Socket.IO
+
+```js
+import http from 'http';
+import {
+  SocketIOSubscriptionServer, // or WebSocketSubscriptionServer
+  JwtCredentialManager,
+  RedisSubscriber,
+} from '@4c/graphql-subscription-server';
+
+const server = http.createServer();
+
+const subscriptionServer = new SocketIOSubscriptionServer({
+  schema,
+  path: '/socket.io/graphql',
+  subscriber: new RedisSubscriber(),
+  hasPermission: (message, credentials) => {
+    authorize(message, credentials);
+  },
+  createCredentialsManager: (req) => new JwtCredentialManager(),
+  createLogger: () => console.debug,
+});
+
+subscriptionServer.attach(server);
+
+server.listen(4000, () => {
+  console.log('server running');
+});
+```

package.json

@@ -19,7 +19,8 @@
     "tdd": "jest --watch",
     "test": "yarn lint && yarn typecheck && jest",
     "testonly": "jest",
-    "typecheck": "tsc --noEmit && tsc -p test --noEmit"
+    "typecheck": "tsc --noEmit && tsc -p test --noEmit",
+    "update-schema": "NODE_ENV=test babel-node ./update-schema.js"
   },
   "husky": {
     "hooks": {
@@ -51,7 +52,7 @@
     "express": "^4.17.1",
     "graphql-ws": "^4.3.2",
     "redis": "^3.0.0",
-    "ws": "^7.4.4"
+    "ws": "^7.4.5"
   },
   "peerDependencies": {
     "graphql": ">=0.12.3",
@@ -71,6 +72,7 @@
     "@4c/tsconfig": "^0.3.1",
     "@babel/cli": "^7.12.10",
     "@babel/core": "^7.12.10",
+    "@babel/node": "^7.14.2",
     "@babel/preset-typescript": "^7.12.7",
     "@types/express": "^4.17.9",
     "@types/jest": "^26.0.19",
@@ -88,13 +90,16 @@
     "eslint-plugin-jsx-a11y": "^6.4.1",
     "eslint-plugin-prettier": "^3.3.0",
     "graphql": "^15.4.0",
+    "graphql-relay": "^0.6.0",
+    "graphql-relay-subscription": "^0.3.1",
     "husky": "^4.3.6",
     "jest": "^26.6.3",
     "lint-staged": "^10.5.3",
     "prettier": "^2.2.1",
     "redis-mock": "^0.55.1",
     "semantic-release": "^17.3.0",
-    "socket.io": "^3.0.4",
+    "socket.io": "^4.1.2",
+    "socket.io-client": "^4.1.2",
     "travis-deploy-once": "^5.0.11",
     "typescript": "^4.1.3",
     "utility-types": "^3.10.0"

src/AuthorizedSocketConnection.ts

@@ -9,13 +9,13 @@ import {
   validate,
 } from 'graphql';
 import { ExecutionResult } from 'graphql/execution/execute';
-import io from 'socket.io';
 
 import * as AsyncUtils from './AsyncUtils';
 import { CredentialsManager } from './CredentialsManager';
 import { CreateLogger, Logger } from './Logger';
 import { Subscriber } from './Subscriber';
 import SubscriptionContext from './SubscriptionContext';
+import { WebSocket } from './types';
 
 export type CreateValidationRules = ({
   query,
@@ -62,7 +62,7 @@ const acknowledge = (cb?: () => void) => {
  * - Rudimentary connection constraints (max connections)
  */
 export default class AuthorizedSocketConnection<TContext, TCredentials> {
-  socket: io.Socket;
+  socket: WebSocket;
 
   config: AuthorizedSocketOptions<TContext, TCredentials>;
 
@@ -74,7 +74,7 @@ export default class AuthorizedSocketConnection<TContext, TCredentials> {
   >;
 
   constructor(
-    socket: io.Socket,
+    socket: WebSocket,
     config: AuthorizedSocketOptions<TContext, TCredentials>,
   ) {
     this.socket = socket;
@@ -83,12 +83,11 @@ export default class AuthorizedSocketConnection<TContext, TCredentials> {
     this.log = config.createLogger('@4c/SubscriptionServer::AuthorizedSocket');
     this.subscriptionContexts = new Map();
 
-    this.socket
-      .on('authenticate', this.handleAuthenticate)
-      .on('subscribe', this.handleSubscribe)
-      .on('unsubscribe', this.handleUnsubscribe)
-      .on('connect', this.handleConnect)
-      .on('disconnect', this.handleDisconnect);
+    this.socket.on('authenticate', this.handleAuthenticate);
+    this.socket.on('subscribe', this.handleSubscribe);
+    this.socket.on('unsubscribe', this.handleUnsubscribe);
+    this.socket.on('connect', this.handleConnect);
+    this.socket.on('disconnect', this.handleDisconnect);
   }
 
   emitError(error: { code: string; data?: any }) {

src/SocketIOSubscriptionServer.ts

@@ -0,0 +1,67 @@
+import { promisify } from 'util';
+
+import express from 'express';
+import type io from 'socket.io';
+
+import SubscriptionServer, {
+  SubscriptionServerConfig,
+} from './SubscriptionServer';
+
+export interface SocketIOSubscriptionServerConfig<TContext, TCredentials>
+  extends SubscriptionServerConfig<TContext, TCredentials> {
+  socketIoServer?: io.Server;
+}
+
+export default class SocketIOSubscriptionServer<
+  TContext,
+  TCredentials
+> extends SubscriptionServer<TContext, TCredentials> {
+  io: io.Server;
+
+  constructor({
+    socketIoServer,
+    ...config
+  }: SocketIOSubscriptionServerConfig<TContext, TCredentials>) {
+    super(config);
+
+    this.io = socketIoServer!;
+    if (!this.io) {
+      // eslint-disable-next-line global-require, @typescript-eslint/no-var-requires
+      const IoServer = require('socket.io').Server;
+      this.io = new IoServer({
+        serveClient: false,
+        path: this.config.path,
+        transports: ['websocket'],
+        allowEIO3: true,
+      });
+    }
+
+    this.io.on('connection', (socket: io.Socket) => {
+      const request = Object.create((express as any).request);
+      Object.assign(request, socket.request);
+      this.opened(
+        {
+          id: socket.id,
+          protocol: 'socket-io',
+          on: socket.on.bind(socket),
+          emit(event: string, data: any) {
+            socket.emit(event, data);
+          },
+          close() {
+            socket.disconnect();
+          },
+        },
+        request,
+      );
+    });
+  }
+
+  attach(httpServer: any) {
+    this.io.attach(httpServer);
+  }
+
+  async close() {
+    // @ts-ignore
+    await promisify((...args) => this.io.close(...args))();
+  }
+}