UCOMmandIt
/
uci-socket
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

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

tls
David Kebler 2018-05-24 12:28:30 -07:00
parent e2e59466be
commit 949bd5b3b1
3 changed files with 52 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
package.json
View File

@ -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": {

10
src/consumer.js
View File

@ -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) {

56
src/socket.js
View File

@ -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