Merge pull request #71 from BackendStack21/ws-support

Introducing WebSocket Support

Author: jkyberneees

.gitignore

@@ -60,4 +60,4 @@ typings/
 # next.js build output
 .next
 
-.DS_Store
+.DS_Store

README.md

@@ -196,6 +196,41 @@ Example output:
 ```
 > NOTE: Please see `docs` configuration entry explained above.
 
+### WebSockets support
+WebSockets proxying is supported since `v3.1.0`. Main considerations:
+- WebSockets middlewares are not supported.
+- WebSocketRoute configuration definition:
+  ```ts
+  interface WebSocketRoute {
+    proxyType: 'websocket';
+    // https://github.com/faye/faye-websocket-node#initialization-options
+    proxyConfig?: {}; 
+    prefix: string;
+    target: string;
+    // https://github.com/faye/faye-websocket-node#subprotocol-negotiation
+    subProtocols?: []; 
+    hooks?: WebSocketHooks;
+  }
+
+  interface WebSocketHooks {
+    onOpen?: (ws: any, searchParams: URLSearchParams) => Promise<void>;
+  }
+  ```
+- The `/` route prefix is considered the default route. 
+
+#### Configuration example:
+```js
+gateway({
+  routes: [{
+    // ... other HTTP or WebSocket routes
+  }, {
+    proxyType: 'websocket',
+    prefix: '/echo',
+    target: 'ws://ws.ifelse.io'
+  }]
+}).start(PORT)
+```
+
 ## Timeouts and Unavailability 
 We can restrict requests timeouts globally or at service level using the `timeout` configuration.  
 

benchmark/websocket/artillery-perf1.yml

@@ -0,0 +1,11 @@
+config:
+  target: "ws://localhost:8080/echo"
+  phases:
+    - duration: 15
+      arrivalRate: 20
+      rampTo: 500
+      name: "Ramping up the load"
+scenarios:
+  - engine: "ws"
+    flow:
+      - send: 'echo me'

benchmark/websocket/echo.js

@@ -0,0 +1,25 @@
+'use strict'
+
+const gateway = require('../../index')
+const WebSocket = require('faye-websocket')
+const http = require('http')
+
+gateway({
+  routes: [{
+    proxyType: 'websocket',
+    prefix: '/echo',
+    target: 'ws://127.0.0.1:3000'
+  }]
+}).start(8080)
+
+const service = http.createServer()
+service.on('upgrade', (request, socket, body) => {
+  if (WebSocket.isWebSocket(request)) {
+    const ws = new WebSocket(request, socket, body)
+
+    ws.on('message', (event) => {
+      ws.send(event.data)
+    })
+  }
+})
+service.listen(3000)

demos/ws-proxy.js

@@ -0,0 +1,21 @@
+'use strict'
+
+const gateway = require('./../index')
+const PORT = process.env.PORT || 8080
+
+gateway({
+  routes: [{
+    // ... other HTTP or WebSocket routes
+  }, {
+    proxyType: 'websocket',
+    prefix: '/echo',
+    target: 'ws://ws.ifelse.io',
+    hooks: {
+      onOpen: (ws, searchParams) => {
+
+      }
+    }
+  }]
+}).start(PORT).then(server => {
+  console.log(`API Gateway listening on ${PORT} port!`)
+})

index.d.ts

@@ -33,6 +33,19 @@ declare namespace fastgateway {
     hooks?: Hooks;
   }
 
+  interface WebSocketRoute {
+    proxyType: 'websocket';
+    proxyConfig?: {}; // https://github.com/faye/faye-websocket-node#initialization-options
+    prefix: string;
+    target: string;
+    subProtocols?: []; // https://github.com/faye/faye-websocket-node#subprotocol-negotiation
+    hooks?: WebSocketHooks;
+  }
+
+  interface WebSocketHooks {
+    onOpen?: (ws: any, searchParams: URLSearchParams) => Promise<void>;
+  }
+
   interface Hooks {
     onRequest?: Function,
     rewriteHeaders?: Function,
@@ -54,7 +67,7 @@ declare namespace fastgateway {
     pathRegex?: string;
     timeout?: number;
     targetOverride?: string;
-    routes: Route[];
+    routes: Route[] | WebSocketRoute[];
   }
 }
 

index.js

@@ -8,6 +8,7 @@ const defaultProxyHandler = (req, res, url, proxy, proxyOpts) => proxy(req, res,
 const DEFAULT_METHODS = require('restana/libs/methods').filter(method => method !== 'all')
 const send = require('@polka/send-type')
 const PROXY_TYPES = ['http', 'lambda']
+const registerWebSocketRoutes = require('./lib/ws-proxy')
 
 const gateway = (opts) => {
   const proxyFactory = opts.proxyFactory || defaultProxyFactory
@@ -17,24 +18,30 @@ const gateway = (opts) => {
     pathRegex: '/*'
   }, opts)
 
-  const server = opts.server || restana(opts.restana)
+  const router = opts.server || restana(opts.restana)
 
   // registering global middlewares
   opts.middlewares.forEach(middleware => {
-    server.use(middleware)
+    router.use(middleware)
   })
 
   // registering services.json
   const services = opts.routes.map(route => ({
     prefix: route.prefix,
     docs: route.docs
   }))
-  server.get('/services.json', (req, res) => {
+  router.get('/services.json', (req, res) => {
     send(res, 200, services)
   })
 
-  // processing routes
-  opts.routes.forEach(route => {
+  // processing websocket routes
+  registerWebSocketRoutes({
+    routes: opts.routes.filter(route => route.proxyType === 'websocket'),
+    server: router.getServer()
+  })
+
+  // processing non-websocket routes
+  opts.routes.filter(route => route.proxyType !== 'websocket').forEach(route => {
     if (undefined === route.prefixRewrite) {
       route.prefixRewrite = ''
     }
@@ -82,13 +89,13 @@ const gateway = (opts) => {
 
     methods.forEach(method => {
       method = method.toLowerCase()
-      if (server[method]) {
-        server[method].apply(server, args)
+      if (router[method]) {
+        router[method].apply(router, args)
       }
     })
   })
 
-  return server
+  return router
 }
 
 const handler = (route, proxy, proxyHandler) => async (req, res, next) => {

lib/default-hooks.js

@@ -5,6 +5,9 @@ const toArray = require('stream-to-array')
 const TRANSFER_ENCODING_HEADER_NAME = 'transfer-encoding'
 
 module.exports = {
+  websocket: {
+    onOpenNoOp (ws, searchParams) { }
+  },
   lambda: {
     onRequestNoOp (req, res) { },
     onResponse (req, res, response) {

lib/ws-proxy.js

@@ -0,0 +1,43 @@
+'use strict'
+
+const WebSocket = require('faye-websocket')
+const { onOpenNoOp } = require('./default-hooks').websocket
+
+module.exports = (config) => {
+  const { routes, server } = config
+
+  const prefix2route = new Map()
+  for (const route of routes) {
+    prefix2route.set(route.prefix, route)
+  }
+
+  server.on('upgrade', async (req, socket, body) => {
+    if (WebSocket.isWebSocket(req)) {
+      const url = new URL('http://fw' + req.url)
+      const prefix = url.pathname || '/'
+
+      if (prefix2route.has(prefix)) {
+        const route = prefix2route.get(prefix)
+        const subProtocols = route.subProtocols || []
+        route.hooks = route.hooks || {}
+        const onOpen = route.hooks.onOpen || onOpenNoOp
+
+        const client = new WebSocket(req, socket, body, subProtocols)
+
+        try {
+          await onOpen(client, url.searchParams)
+
+          const target = route.target + '?' + url.searchParams.toString()
+          const remote = new WebSocket.Client(target, subProtocols, route.proxyConfig)
+
+          client.pipe(remote)
+          remote.pipe(client)
+        } catch (err) {
+          client.close(err.closeEventCode || 4500, err.message)
+        }
+      } else {
+        socket.end()
+      }
+    }
+  })
+}

package.json

@@ -7,7 +7,8 @@
   "scripts": {
     "test": "nyc mocha test/*.test.js",
     "format": "npx standard --fix",
-    "lint": "npx standard"
+    "lint": "npx standard",
+    "ws-bench": "npx artillery run benchmark/websocket/artillery-perf1.yml"
   },
   "repository": {
     "type": "git",
@@ -42,12 +43,14 @@
   ],
   "devDependencies": {
     "@types/express": "^4.17.11",
-    "aws-sdk": "^2.1023.0",
+    "artillery": "^2.0.0-6",
+    "aws-sdk": "^2.1037.0",
     "chai": "^4.3.4",
     "cors": "^2.8.5",
     "express": "^4.17.1",
     "express-jwt": "^6.1.0",
     "express-rate-limit": "^5.5.1",
+    "faye-websocket": "^0.11.4",
     "fg-multiple-hooks": "^1.3.0",
     "helmet": "^4.6.0",
     "http-lambda-proxy": "^1.1.4",