From 949bd5b3b153fa891209a15ba6aa92b7946d0725 Mon Sep 17 00:00:00 2001 From: David Kebler Date: Thu, 24 May 2018 12:28:30 -0700 Subject: [PATCH] add client tracking on server side and push notification processin to both. Pushed packets have _header.id of 'pushed' and server has new push method --- package.json | 2 +- src/consumer.js | 10 +++++++++ src/socket.js | 56 ++++++++++++++++++++++++++++++++++++------------- 3 files changed, 52 insertions(+), 16 deletions(-) diff --git a/package.json b/package.json index 34f40cd..1ade8cd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@uci/socket", - "version": "0.1.2", + "version": "0.1.3", "description": "JSON packet intra(named)/inter(TCP) host communication over socket", "main": "src", "scripts": { diff --git a/src/consumer.js b/src/consumer.js index 34c4c2f..ea258e2 100644 --- a/src/consumer.js +++ b/src/consumer.js @@ -138,8 +138,18 @@ export default class Consumer extends Socket { async _listen () { log.info('listening for incoming packets from socket') + // listen for pushed packets + this.on('pushed',async function(packet){ + // TODO do some extra security here? + let res = await this._packetProcess(packet) + if (!res) { // if process was not promise returning like just logged to console + log.warn('consumer function was not promise returning - resolving unprocessed') + } + }) + // listen on socket stream this.on('data', this.stream.onData) this.stream.on('message', messageProcess.bind(this)) + async function messageProcess (packet) { // console.log('incoming packet from socket',packet) if (packet._handshake) { diff --git a/src/socket.js b/src/socket.js index bfe9c4c..ccf8df9 100644 --- a/src/socket.js +++ b/src/socket.js @@ -26,6 +26,8 @@ export default class Socket extends Server { if (typeof opts.path === 'boolean') opts.path = path.join(DEFAULT_PIPE_DIR,DEFAULT_SOCKET_NAME ) if (path.dirname(opts.path)==='.') opts.path = path.join(DEFAULT_PIPE_DIR,opts.path ) } + this.clientTracking = opts.clientTracking || true + this.clients = [] // track consumers (i.e. clients) this.opts = opts // for use to recover from selected errors //self bindings this._listen = this._listen.bind(this) @@ -83,28 +85,41 @@ export default class Socket extends Server { if (err) return err // this gets called for each client connection and is unique to each this.on('connection', async (socket) => { - let write = this._write.bind(socket) const stream = new JSONStream() - log.info('new consumer connecting sending handshake') - write(await stream.serialize({'_handshake':true})) + socket.stream = stream // need this to track clients + let send = this._send.bind(socket) + if (this.clientTracking) this.clients.push(socket) + // TODO add 'close' listener to socket to remove from this.clients + log.info('new consumer connecting') + log.info(await send(await stream.serialize({'_handshake':true}))) socket.on('data', stream.onData) // TODO need to start error listener for stream so errors can be processed - stream.on('message', messageProcess.bind(this)) - async function messageProcess (packet) { - //console.log(' incoming packet on socket side',packet) - let res = await this._packetProcess(clone(packet)) - if (Object.keys(res).length === 0) res = { error: 'socket packet command function likely did not return a promise', packet:packet} - res._header = clone(packet._header,false) //make sure return packet has header with id in case it was removed in processing - delete packet._header // remove before adding to response header as request + stream.on('message', messageProcess.bind(this,socket)) + + async function messageProcess (client, packet) { + log.info({packet:packet},'incoming packet on socket side') + let res = {} + if (this.clientTracking && packet.clientID) { + client.ID = packet.clientID + res.cmd='ackID' + } + else { + res = await this._packetProcess(clone(packet)) || {} + if (Object.keys(res).length === 0) res = { error: 'socket packet command function likely did not return a promise', packet:packet} + } + if (packet) { + res._header = clone(packet._header,false) || {} //make sure return packet has header with id in case it was removed in processing + delete packet._header // remove before adding to response header as request + } else res._header = {} res._header.request = clone(packet,false) res._header.responder = {name:this.name,instanceID:this.id} - res._header.socket = this._connectionKey + res._header.socket = this.address() if (!res.cmd) res.cmd = 'reply' // by default return command is 'reply' let [err, ser] = await btc(stream.serialize)(res) if (err) ser = await stream.serialize({ error: 'was not able to serialze the res packet', err:err, _header:{id:res._header.id}}) - // console.log('serialized ready for write',ser) - log.info(await write(ser)) - } + log.info(await send(ser)) + } // end process message + }) // end connecttion consumer log.info({opts: this.opts},'socket created') return res @@ -127,7 +142,7 @@ export default class Socket extends Server { } // must have a consumer socket bound to use - async _write(packet) { + async _send(packet) { // timeout already set if sockect can't be drained in 10 secs return new Promise(resolve => { const cb = () => resolve('packet written to socket stream') @@ -139,4 +154,15 @@ export default class Socket extends Server { }) } + async push (packet,id) { + packet._header = { id:'pushed'} + this.clients.forEach(async (client) => { + if (client.writable) { + let [err, ser] = await btc(client.stream.serialize)(packet) + if (err) ser = await client.stream.serialize({ error: 'was not able to serialze the res packet', err:err, _header:{id:packet._header.id}}) + if (!id || id ===client.ID ) await this._send.bind(client)(ser) + } + }) + } + } // end class