refactored connected and added auto reconnect ability like the unix/tcp socket including authentication and per consumer ping

added in consumer ping

examples/server.js

@@ -0,0 +1,86 @@
+import Socket from '../src'
+
+const ANONYMOUS = true
+const PORT = 8090
+
+function packetProcess (packet) {
+
+  const cmd = {
+    ack: function(packet){
+      return new Promise( async (resolve) => {
+        packet.ack = true
+        packet.sender= packet.sender || (packet._header ? packet._header.sender.name : 'unknown')
+        packet.msg=`this is acknlowdgement that ${packet.sender} ack was received`
+        this.push(packet)  // push to all active socket servers
+        return resolve(packet) // this goes back to sender
+      })
+    },
+    switch:{
+      status: function(packet){
+        return new Promise( async (resolve) => {
+          packet.cmd='switch/status'
+          packet.state=this.switches[packet.id-1]
+          packet.sender= packet.sender || (packet._header ? packet._header.sender.name : 'unknown')
+          this.push(packet)  // push to all active socket servers
+          let res = { response:'status pushed on to all clients'}
+          return resolve(res) // this goes back to sender
+        })
+      },
+      on: function(packet){
+        return new Promise( async (resolve) => {
+          packet.cmd='switch/status'
+          packet.state='on'
+          this.switches[packet.id-1] = 'on'
+          packet.sender= packet.sender || (packet._header ? packet._header.sender.name : 'unknown')
+          this.push(packet)  // push to all active socket servers
+          let res = { response:'status change - pushed on to all clients', id:packet.id}
+          return resolve(res) // this goes back to sender
+        })
+      },
+      off: function(packet){
+        return new Promise( async (resolve) => {
+          packet.cmd='switch/status'
+          packet.state='off'
+          this.switches[packet.id-1] = 'off'
+          packet.sender= packet.sender || (packet._header ? packet._header.sender.name : 'unknown')
+          this.push(packet)  // push to all active socket servers
+          let res = { response:'status change - pushed off to all clients'}
+          return resolve(res)  // this goes back to sender
+        })
+      },
+      toggle: function(packet){
+        return new Promise( async (resolve) => {
+          this.switches[packet.id-1] = this.switches[packet.id-1]==='on' ? 'off' : 'on'
+          packet.state = this.switches[packet.id-1]
+          packet.cmd ='switch/status'
+          packet.sender= packet.sender || (packet._header ? packet._header.sender.name : 'unknown')
+          this.push(packet)  // push to all active socket servers
+          let res = { response:`status change - pushed toggled state of ${packet.state} to all clients`}
+          return resolve(res) // this goes back to sender
+        })
+      }
+    }
+  }
+  console.log('this in packet processor',packet.cmd)
+  let cmdProps = packet.cmd.split('/')
+  let func = cmd[cmdProps[0]]
+  console.log(func)
+  if (cmdProps[1]) func = func[cmdProps[1]]
+  console.log(func)
+  if (typeof func === 'function')  return func.call(this,packet)
+  else {
+    packet.error='no function for command'
+    return packet
+  }
+}
+
+
+// let test = new Test()
+let test = new Socket({ port: PORT, allowAnonymous:ANONYMOUS })
+test.switches = ['off','off','off','off']  // database to hold switch states
+test.registerPacketProcessor(packetProcess)
+;(async () => {
+  console.log(await test.create())
+})().catch(err => {
+  console.error('FATAL: UNABLE TO START SYSTEM!\n', err)
+})

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@uci/websocket",
-  "version": "0.4.1",
+  "version": "0.3.8",
   "description": "JSON packet host websocket server",
   "main": "src",
   "scripts": {
@@ -33,16 +33,15 @@
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",
     "esm": "^3.2.25",
-    "mocha": "^8.0.1",
-    "nodemon": "^2.0.4"
+    "mocha": "^6.2.0",
+    "nodemon": "^1.19.1"
   },
   "dependencies": {
-    "@uci-utils/logger": "0.0.18",
+    "@uci-utils/logger": "0.0.15",
     "await-to-js": "^2.1.1",
     "better-try-catch": "^0.6.2",
     "clone": "^2.1.2",
     "death": "^1.1.0",
-    "p-reflect": "^2.1.0",
-    "ws": "^7.3.1"
+    "ws": "^7.1.2"
   }
 }

src/socket.js

@@ -1,7 +1,6 @@
 import { Server as WSS } from 'ws'
 import { Server } from 'http'
 import btc from 'better-try-catch'
-import pReflect from 'p-reflect'
 import { promisify } from 'util'
 import _ON_DEATH from 'death' //this is intentionally ugly
 import clone from 'clone'
@@ -9,6 +8,10 @@ import clone from 'clone'
 import logger from '@uci-utils/logger'
 let log = {}
 
+
+
+
+
 /**
  * Socket - Description
  * @extends Server
@@ -34,17 +37,14 @@ class Socket extends Server {
     this.pingInterval = opts.pingInterval === false ? opts.pingInterval : (opts.pingInterval * 1000 || 5000)
     this.nextconsumerID = 0 // incrementer for default initial consumer ID
     this.consumers = new Map()
-    this.conPackets = opts.conPackets || [opts.conPacket]
     this.errors = []
     this.errorCount = 0
     this.create = this.create.bind(this)
     this._destroy = this._destroy.bind(this)
-    this._listening=false
     this.authenticateconsumer = this.authenticateconsumer.bind(this)
     this._authenticate = this._authenticate.bind(this)
     this.close = promisify(this.close).bind(this)
     log = logger({
-      package:'@uci/websocket',
       file: 'src/socket.js',
       class: 'Socket',
       name: 'websocket',
@@ -52,17 +52,13 @@ class Socket extends Server {
     })
   } // end constructor
 
-  get active() {
-    return this._listening
-  }
-
   /**
    * create - Description
    *
    * @returns {type} Description
    */
   async create() {
-    this.emit('socket',{state:'creating', msg:'creating socket for consumers to connect'})
+    // return Promise.resolve('what')
     return new Promise((resolve, reject) => {
       _ON_DEATH(async () => {
         log.error({method:'create', line:51, msg:'\nhe\'s dead jim'})
@@ -86,28 +82,12 @@ class Socket extends Server {
         this.on('error', err => {
           this.errorCount +=1 // log errors here
           this.errors.push(err)
-          if(this.errorCount>2) {
-            this.emit('log', {level:'warn',msg:'something bad maybe going on, 3 errors', errors:this.errors})
-            this.emit('socket',{state:'error', msg:'2 to 5 socket errors', errors:this.errors})
-          }
-          if(this.errorCount>5) {
-            let errors = {level:'fatal',msg:'something fatal is going on, 6 errors', errors:this.errors}
-            log.fatal(errors)
-            this._listening=false
-            this.close(() => {
-              this.emit('socket',{state:'offline', msg:'too many socket errors no longer listening for consumers to connect'})
-            })
-            this.emit('log', {active:this.active})
-            this.emit('log', errors)
-          }
+          if(this.errorCount>2) this.emit('warn', {msg:'something bad maybe going on, 3 errors', errors:this.errors})
+          if(this.errorCount>5) this.emit('fatal', {msg:'something fatal is going on, 6 errors', errors:this.errors})
         })
         this.wss = new WSS({server:this})
         this.wss.on('error', err => {this.emit('error', err)}) // bubble up errors
         this.wss.on('connection', this._connectionHandler.bind(this))
-        this._listening=true
-        let msg = `socket ready and listening ${typeof this.address() ==='string' ? `at ${this.address()}` : `on port ${this.address().port}`}`
-        this.emit('log',{active:this.active})
-        this.emit('socket',{state:'listening', msg:msg})
         resolve(`websocket ready and listening at ${this.address().address}:${this.address().port}`)
       })
       super.listen(this.opts)
@@ -161,43 +141,20 @@ class Socket extends Server {
    * @param {<type>}	id - this is the parameter id
    *
    */
-  async push(packet={},opts={}) {
+  async push(packet={},id) {
+    packet._header = {id: id || 'pushed'}
     if (this.consumers.size > 0) {
-      packet._header = {id: opts.packetId || 'pushed'}
-      let consumers = []
-      if ( opts.consumers || opts.consumer ) {
-        if (opts.consumer) opts.consumers = [opts.consumer]
-        consumers = Array.from(this.consumers).filter(([sid,consumer]) =>
-          opts.consumers.some(con=> {
-            console.log(consumer.sid,consumer.data,con)
-            return (
-              con === ( (consumer.data ||{}).name || (consumer.data ||{}).id ) ||
-               con.sid=== sid ||
-               con.name === (consumer.data ||{}).name  ||
-               con.id === (consumer.data ||{}).id
-            )
-          }
-          )
-        ).map(con=>con[1])
-        // console.log('custom consumers',consumers.length)
-      } else consumers = Array.from(this.consumers.values())
-
-      consumers = consumers.filter(consumer=>consumer.writable||consumer.readyState===1)
-
-      const send = consumer => this._send(consumer,packet)
-      const res = await Promise.all(consumers.map(send).map(pReflect))
-
-      const success = res.filter(result => result.isFulfilled).map((result,index) => [consumers[index].name,result.value])
-      const errors =res.filter(result => result.isRejected).map((result,index) => [consumers[index].name,result.reason])
-      this.emit('log',{level:errors.length? 'error': packet._header.id==='ping'?'trace':'debug', msg:'packet was pushed', socket:this.name||this.id, errors:errors, packet:packet, success:success, headerId:packet._header.id})
+      log.debug({method:'push', line:142, id:packet._header.id, opts: this.opts, packet: packet, msg:'pushing a packet to all connected consumers'})
+      this.consumers.forEach(async consumer => {
+        this._seend(consumer,packet)
+      })
     } else {
-      this.emit('log',{level:'debug', msg:'no connected consumers packet push ignored', packet:packet})
+      log.debug({method:'push', line:165, id:packet._header.id, opts: this.opts, packet: packet, msg:'no connected consumers, push ignored'})
     }
   }
 
   removeconsumer (id) {
     let consumer = this.consumers.get(id)
-    this.emit('connection:consumer',{state:'disconnected', name:consumer.name})
     clearInterval(consumer._ping)
     consumer.removeAllListeners()
     log.warn({msg:`consumer ${id}:${consumer.name} removed from server tracking`, id:id, name:consumer.name, curconsumerCount:this.consumers.size})
@@ -235,7 +192,7 @@ class Socket extends Server {
           }
           else {
             log.info({msg:'consumer authenticated successfuly', consumer:consumer.name, consumer_id:consumer.id})
-            if (this.allowAnonymous) log.warn({msg:'web consumer connected anonymously', consumer:consumer.name, consumer_id:consumer.id})
+            if (this.allowAnonymous) log.warn({msg:'consumer connected anonymously', consumer:consumer.name, consumer_id:consumer.id})
             resolve(consumer.authenticated)
           }
         }
@@ -287,15 +244,12 @@ class Socket extends Server {
       }
     }
 
-    if (this.conPackets.length) {
-      for (let packet of this.conPackets) {
-        packet._header = {type:'on connection packet', id: 'pushed'}
-        await this._send(consumer,packet) // send a packet command on to consumer on connection
-      }
+    if (this.opts.conPacket) {
+      this.opts.conPacket._header = { id: 'pushed' }
+      log.debug({method:'_listen', line:171, conPacket: this.opts.conPacket, msg:'pushing a preset command to just connected consumer'})
+      this._send(consumer,this.opts.conPacket) // send a packet command on to consumer on connection
     }
 
-    this.emit('connection:consumer',{state:'connected', name:consumer.name, id:consumer.id})
-
     consumer._ping = setInterval( () => {
       consumer.ping(JSON.stringify({pingInterval:this.pingInterval}))
       this._send(consumer,{_header:{id:'ping'},pingInterval:this.pingInterval})
@@ -311,7 +265,7 @@ class Socket extends Server {
         res = { error: 'Could not JSON parse packet', packet:strPacket }
       }
       else {
-        log.debug({method:'_listen', line:266, packet:packet, msg:'parsed packet'})
+        log.warn({method:'_listen', line:266, packet:packet, msg:'parsed packet'})
         res = (await this._packetProcess(clone(packet))) || {}
         if (Object.keys(res).length === 0)
           res = {