basic websocket server listening

Author: sevenbitbyte

examples/test-service-host.js

@@ -37,7 +37,7 @@ async function main(){
 
   const service = new ExampleService({ name: '@dataparty/example', version: '0.0.1' })
 
-  const build = await service.compile(Path.join(__dirname,'../dataparty'), true)
+  const build = await service.compile(Path.join(__dirname,'/dataparty'), true)
 
   debug('built', Object.keys(build))
 
@@ -46,7 +46,11 @@ async function main(){
     sendFullErrors: true
   })
 
-  const host = new Dataparty.ServiceHost({runner, trust_proxy: true})
+  const host = new Dataparty.ServiceHost({
+    runner,
+    trust_proxy: false,
+    wsEnabled: true
+  })
 
   await party.start()
   await runner.start()

package.json

@@ -102,6 +102,7 @@
     "uuidv4": "^6.2.12",
     "vm2": "^3.9.2",
     "websocket": "github:sevenbitbyte/WebSocket-Node#parcel-build",
+    "ws": "^8.11.0",
     "zangodb": "github:sevenbitbyte/zangodb#hash-patch"
   },
   "devDependencies": {

src/service/service-host-websocket.js

@@ -0,0 +1,99 @@
+const {URL} = require('url')
+const debug = require('debug')('dataparty.service-host-websocket')
+
+const ws = require('ws')
+const WebSocketServer = ws.WebSocketServer
+
+const WATCHDOG_INTERVAL = 30000
+
+class ServiceHostWebsocket{
+
+  constructor({trust_proxy, port, path, runner, wsSettings}){
+    this.port = port
+    this.path = path || '/ws'
+    this.runner = runner
+    this.trust_proxy = trust_proxy
+    this.wsSettings = wsSettings || {}
+
+    this.ws = null
+  }
+
+  start(server){
+
+    debug('start')
+
+    let settings = {}
+
+    if(!server){
+      settings = { port: this.port, ...this.wsSettings }
+      
+    } else {
+      settings = { noServer: true, ...this.wsSettings }
+    }
+
+    debug('\t','settings', settings)
+
+    this.ws = new WebSocketServer(settings)
+
+    this.ws.on('connection', this.handleConnection.bind(this))
+
+    server.on('upgrade', this.handleUpgrade.bind(this))
+
+    this.watchdog = setInterval(this.checkClients.bind(this), WATCHDOG_INTERVAL) 
+  }
+
+  handleUpgrade(request, socket, head){
+
+    debug('handleUpgrade', request.url)
+
+    if(request.url == this.path){
+      this.doUpgrade(request, socket, head)
+    } else {
+      socket.destroy()
+    }
+  }
+
+  doUpgrade(request, socket, head){
+    debug('doUpgrade')
+    this.ws.handleUpgrade(request, socket, head, (conn)=>{
+      debug('head', head)
+      this.ws.emit('connection', conn, request)
+    })
+  }
+
+  getConnectionIp(req){
+    if(this.trust_proxy){
+      const ip = req.headers['x-forwarded-for'].split(',')[0].trim();
+
+      return ip
+    }
+
+    return req.socket.remoteAddress
+  }
+
+  handleConnection(conn, req){
+   conn.ip = this.getConnectionIp(req)
+    
+    debug('handleConnection - ', conn.ip)
+
+    conn.isAlive = true
+    conn.on('pong', ()=>{
+      conn.isAlive = true
+    })
+  }
+
+  checkClients(){
+    debug('checkClients')
+    this.ws.clients.forEach( (conn)=>{
+      if (conn.isAlive === false){
+        debug('\t','terminating client')
+        return conn.terminate()
+      }
+  
+      conn.isAlive = false
+      conn.ping()
+    });
+  }
+}
+
+module.exports = ServiceHostWebsocket

src/service/service-host.js

@@ -9,6 +9,8 @@ const bodyParser = require('body-parser')
 const expressListRoutes = require('express-list-routes')
 const debug = require('debug')('dataparty.service-host')
 
+const ServiceHostWebsocket = require('./service-host-websocket')
+
 const Pify = async (p)=>{
   return await p
 }
@@ -22,6 +24,8 @@ class ServiceHost {
     cors = {},
     trust_proxy = false,
     listenUri = 'http://0.0.0.0:4001',
+    wsEnabled = true,
+    wsPort = null,
     runner
   }={}){
     this.apiApp = express()
@@ -45,11 +49,20 @@ class ServiceHost {
     this.errorHandlerTimer = null
 
     this.apiServerUri = new URL(listenUri)
+
+    if(wsEnabled){
+      this.wsServer = new ServiceHostWebsocket({
+        trust_proxy,
+        port: wsPort,
+        runner: this.runner
+      })
+    }
+
   }
 
   async start(){
 
-    debug('starting server')
+    debug('starting server', this.apiServerUri.toString())
 
     if(this.apiServer==null){
       debug('adding default endpoints')
@@ -81,7 +94,6 @@ class ServiceHost {
 
     }
 
-    debug('server listening', this.apiServerUri.toString())
 
     await new Promise((resolve,reject)=>{
       this.apiServer.listen(listenPort, listenHost, resolve)
@@ -94,6 +106,11 @@ class ServiceHost {
 
     debug('server listening')
     debug('address', this.apiServer.address())
+
+    if(this.wsServer && this.apiServer){
+      debug('starting websocket')
+      this.wsServer.start(this.apiServer)
+    }
   }
 
   async stop(){