diff --git a/examples/client.js b/examples/client.js index b2467b4..0dd421f 100644 --- a/examples/client.js +++ b/examples/client.js @@ -70,6 +70,12 @@ client.on('status', event => { console.log('===============') } + setTimeout(()=> { + console.log('=============Consumer now Offline=============') + process.exit(1) + } + ,120000) + })().catch(err => { console.error('FATAL: UNABLE TO START SYSTEM!\n',err) }) diff --git a/examples/server.js b/examples/server.js index aa29dd6..e0db629 100644 --- a/examples/server.js +++ b/examples/server.js @@ -38,6 +38,7 @@ class Test extends Socket { // } // const PATH = '/opt/bogus/socket' const PATH = true +const PUSHES = 3 // options.conPacket = {cmd:'onconnect', data:'this is a packet data sent consumer after handshake/authentification'} const TOKENS = ['cheetos'] let test = new Test({path:PATH, tokens:TOKENS}) @@ -72,14 +73,14 @@ test.registerPacketProcessor(processor) let count = 0 const push = setInterval( () => { count++ - test.push({cmd:'pushed', count:count, status:`some pushed data ${count}`}) - if (count >3) { + test.push({cmd:'pushed', count:count, status:`pushing some data ${count} of ${PUSHES}`}) + if (count >PUSHES) { clearInterval(push) - test.push({cmd:'pushed',status:'now will simulate server going offline by stopping to send ping for 10 seconds'}) + test.push({cmd:'pushed',status:'now will simulate server going offline by stopping to send ping for 15 seconds'}) test.disablePing() setTimeout( () => { test.enablePing() - },10000) + },15000) } },3000) diff --git a/package.json b/package.json index 5648b0a..65f867e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@uci/socket", - "version": "0.2.28", + "version": "0.2.29", "description": "JSON packet intra(named)/inter(TCP) host communication over socket", "main": "src", "scripts": { @@ -16,8 +16,7 @@ "client": "node -r esm examples/client", "devc": "UCI_ENV=dev ./node_modules/.bin/nodemon -r esm examples/client", "devc:token": "UCI_CLIENT_TOKEN='cheetos' npm run devc", - "devc:debug": "UCI_LOG_LEVEL=debug npm run devc", - "c2": "node -r esm examples/client2" + "devc:debug": "UCI_LOG_LEVEL=debug npm run devc" }, "author": "David Kebler", "license": "MIT", diff --git a/src/consumer.js b/src/consumer.js index cf47f22..179b16b 100644 --- a/src/consumer.js +++ b/src/consumer.js @@ -46,10 +46,13 @@ class SocketConsumer extends Socket { opts.path = path.join(DEFAULT_PIPE_DIR, opts.path) } this.opts = opts + this._data = {id:this.id, name:opts.name||this.id} + Object.assign(this._data,opts.data||{}) // holds consumer specific data that will be passed to server in header and on connection // default is keepAlive true, must set to false to explicitly disable // if keepAlive is true then consumer will also be reconnecting consumer + // initTimeout > 4 means socketInit will return a promise this.initTimeout = opts.initTimeout > 4 ? opts.initTimeout * 1000 : null - this.pingFailedTimeout = opts.pingFailedTimeout * 1000 || 10000 + this.pingFailedTimeout = opts.pingFailedTimeout * 1000 || 5000 this.reconnectLimit = opts.reconnectLimit || 0 this.retryWait = opts.retryWait==null ? 5000 : opts.retryWait * 1000 this.heartBeat = !!process.env.HEARTBEAT || opts.heartBeat @@ -116,7 +119,6 @@ class SocketConsumer extends Socket { , this.initTimeout) const successHandler = (ev) => { - console.log('initial success', ev.state) if (ev.state === 'connected') { clearTimeout(initTimeout) this.removeListener('connection:socket',successHandler) @@ -125,7 +127,7 @@ class SocketConsumer extends Socket { } } - this.on('connection:socket',successHandler) + this.on('connection:socket',successHandler.bind(this)) }) } @@ -174,7 +176,7 @@ class SocketConsumer extends Socket { id: Math.random() .toString() .slice(2), // need this for when multiple sends for different consumers use same packet instanceack - sender: { name: this.name, instanceID: this.id }, + sender: { data: this._data, instanceID: this.id}, path: this.opts.path, port: this.opts.port, host: this.opts.host @@ -286,7 +288,7 @@ async function handshake (packet) { this.notify('handshake') let authPacket = this._authenticate() || {} authPacket._authenticate = true - authPacket.clientName = this.id + authPacket.data = this._data let res = await this._authenticateSend(authPacket) this.stream.removeAllListeners('message') clearTimeout(this._doneAuthenticate) @@ -331,7 +333,7 @@ function messageHandler(packet) { // assume all errors are fatal and the socket needs to be disconnected/reconnected async function errorHandler (err) { this.disconnect() - let msg = `error, socket has been disconnected, trying reconnect in ${this.retryWait/1000} secs` + let msg = `error, this consumer is disconnected, trying reconnect in ${this.retryWait/1000} secs` this.notify('error',{error:err,msg:msg}) this.log('error',msg) this._errorRetry = setTimeout(this.reconnect.bind(this),this.retryWait) diff --git a/src/socket-class.js b/src/socket-class.js index d731f34..636965d 100644 --- a/src/socket-class.js +++ b/src/socket-class.js @@ -39,7 +39,7 @@ export default function socketClass(Server) { * @param {String} options.host a tcp host name nornally not used as 0.0.0.0 is set by default * @param {String} options.port a tcp * @param {String | Boolean} options.path xeither full path to where socket should be created or if just 'true' then use default - * @param {Boolean} options.clientTracking track connected clients for push notifications - default: true + * @param {Boolean} options.consumerTracking track connected consumers for push notifications - default: true * @param {Object} options.conPacket A json operson's property * */ @@ -61,14 +61,14 @@ export default function socketClass(Server) { this.allowAnonymous = (!opts.tokens || !!process.env.UCI_ANON || opts.allowAnonymous) ? true : false this.tokens = opts.tokens || [] this.keepAlive = 'keepAlive' in opts ? opts.keepAlive : true - this.pingInterval = opts.pingInterval === false ? opts.pingInterval : (opts.pingInterval * 1000 || 5000) - this.clients = [] // track consumers (i.e. clients) TODO use a Map - this.nextClientID = 0 // incrementer for default initial client ID + this.pingInterval = opts.pingInterval === false ? 0 : (opts.pingInterval * 1000 || 5000) + this.consumers = new Map() // track consumers (i.e. clients) TODO use a Map + this.nextConsumerID = 0 // incrementer for default initial consumer ID this.opts = opts // for use to recover from selected errors this.errorCount = 0 //self bindings this.create = this.create.bind(this) - this.authenticateClient = this.authenticateClient.bind(this) + this.authenticateConsumer = this.authenticateConsumer.bind(this) this._authenticate = this._authenticate.bind(this) this.close = promisify(this.close).bind(this) log = logger({ @@ -141,11 +141,12 @@ export default function socketClass(Server) { this.emit('log', errors) } }) - let obj = {method:'create', line:54, msg:'socket server created and listening at', address:this.address()} + let msg = `socket ready and listening ${typeof this.address() ==='string' ? `at ${this.address()}` : `on port ${this.address().port}`}` + let obj = {method:'create', line:54, msg:msg} log.info(obj) this.on('connection', this._connectionHandler.bind(this)) this.emit('log:',) - resolve(`socket ready and listening at ${this.address().address}:${this.address().port}`) + resolve(msg) }) super.listen(this.opts) @@ -167,7 +168,7 @@ export default function socketClass(Server) { enablePing () { if (this.pingInterval > 499) { this._ping = setInterval( async () =>{ - if (this.clients.length > 0) this.push({pingInterval:this.pingInterval},'ping') + if (this.consumers.size > 0) this.push({pingInterval:this.pingInterval},'ping') },this.pingInterval) } } @@ -212,7 +213,7 @@ export default function socketClass(Server) { /** - * push - pushes a supplied UCI object packet to all connected clients + * push - pushes a supplied UCI object packet to all connected consumers * * @param {object} packet Description * @param {string} id the header id string of the pushed packet, default: 'pushed' @@ -220,12 +221,12 @@ export default function socketClass(Server) { */ async push(packet={},id) { packet._header = {id: id || 'pushed'} - if (this.clients.length > 0) { + if (this.consumers.size > 0) { log.trace({method:'push', line:142, id:packet._header.id, opts: this.opts, packet: packet, msg:'pushing a packet to all connected consumers'}) // TODO should do a map and single promise - this.clients.forEach(async client => { - if (client.writable) { - let [err] = await btc(this._send)(client,packet) + this.consumers.forEach(async consumer => { + if (consumer.writable) { + let [err] = await btc(this._send)(consumer,packet) if (err) log.error({msg:err, error:err}) } }) @@ -234,56 +235,47 @@ export default function socketClass(Server) { } } - // TODO won't need this if moving to a Map - getClientIndex(sid) { - return this.clients.findIndex(client => {return client.sid === sid }) + + removeConsumer (sid) { + let consumer=this.consumers.get(sid) + this.emit('log',{level:'info', msg:'a consumer disconnected', consumer:consumer.data, sid:consumer.sid}) + this.emit('connection:consumer',{state:'disconnected', msg:'a consumer disconnected', consumer:consumer.data, sid:consumer.sid}) + consumer.removeAllListeners() + consumer.stream.removeAllListeners() + this.consumers.delete(sid) + log.warn({msg:'consumer removed from tracking',sid:sid, curConsumerCount:this.consumers.size}) } - getClient(sid) { - return this.clients[this._getClientIndex(sid)] - } - - removeClient (sid) { - let index = this.getClientIndex(sid) - let client=this.clients[index] - this.emit('log',{level:'info', msg:'a consumer disconnected', name:client.name, id:client.id}) - this.emit('connection:consumer',{state:'disconnected', msg:'a consumer disconnected', name:client.name, id:client.id}) - client.removeAllListeners() - client.stream.removeAllListeners() - this.clients.splice(index,1) - log.warn({msg:'consumer removed from tracking',sid:sid, curClientCount:this.clients.length}) - } - - async authenticateClient(client) { + async authenticateConsumer(consumer) { return new Promise(async (resolve, reject) => { // when consumer gets the handshake they must follow with authentication - client.stream.on('message', authenticate.bind(this,client)) - let [err] = await btc(this._send)(client,{_handshake: true, sid:client.sid}) + consumer.stream.on('message', authenticate.bind(this,consumer)) + let [err] = await btc(this._send)(consumer,{_handshake: true, sid:consumer.sid}) if (err) { log.error({msg:'error in handshake send', error:err}) reject(err) } - async function authenticate (client,packet) { - log.debug({msg:`authentication packet from client ${client.name}:${client.id}:${client.sid}`, packet:packet}) - client.stream.removeAllListeners('message') - if (!packet._authenticate) reject('first client packet was not authentication') + async function authenticate (consumer,packet) { + log.debug({msg:`authentication packet from consumer ${consumer.name}:${consumer.id}:${consumer.sid}`, packet:packet}) + consumer.stream.removeAllListeners('message') + if (!packet._authenticate) reject('first consumer packet was not authentication') else { let [err, res] = await btc(this._authenticate)(packet) - client.authenticated = this.allowAnonymous ? 'anonymous' : (err ? false : res) - client.name = packet.clientName - packet.authenticated = client.authenticated + consumer.authenticated = this.allowAnonymous ? 'anonymous' : (err ? false : res) + consumer.data = packet.data + packet.authenticated = consumer.authenticated packet.reason = err || null - log.debug({msg:'sending authorization result to client', packet:packet}) - await this._send(client,packet) // send either way + log.debug({msg:'sending authorization result to consumer', packet:packet}) + await this._send(consumer,packet) // send either way if (err && !this.allowAnonymous) { - log.info({msg:'client authentication failed', client:client.name, client_sid:client.sid, reason:err}) + log.info({msg:`consumer ${consumer.data.name} authentication failed`, consumer:consumer.data, consumer_sid:consumer.sid, reason:err}) reject(packet.reason) } else { - log.info({msg:'client authenticated successfuly', client:client.name}) - if (this.allowAnonymous) log.warn({msg:'socket consumer connected anonymously', consumer:client.name}) - resolve(client.authenticated) + log.info({msg:`consumer ${consumer.data.name} authenticated successfuly`, consumer:consumer.data}) + if (this.allowAnonymous) log.warn({msg:`consumer ${consumer.data.name}, connected anonymously`, consumer:consumer.data}) + resolve(consumer.authenticated) } } } @@ -299,29 +291,30 @@ export default function socketClass(Server) { return false } - // default authenticator - reject value should be reason which is returned to client + // default authenticator - reject value should be reason which is returned to consumer async _authenticate (packet) { if (!this._validateToken(packet.token)) return Promise.reject('invalid token') return true } - // async _connectionHandler({consumer, server}) { // this gets called for each client connection and is unique to - async _connectionHandler(consumer) { // this gets called for each client connection and is unique to each + // async _connectionHandler({consumer, server}) { // this gets called for each consumer connection and is unique to + async _connectionHandler(consumer) { // this gets called for each consumer connection and is unique to each const stream = new JSONStream() consumer.stream = stream + consumer.data = {} consumer.connected = true // add listeners consumer.on('error', (err) => { - log.error({msg:'client connection error',error:err}) + log.error({msg:'consumer connection error',error:err}) // TODO do more handling than just logging }) consumer.on('end', (err) => { - log.error({msg:'consumer connected ended',error:err}) - if (consumer.sid) this.removeClient(consumer.sid) + log.error({msg:`'consumer connection ended: ${consumer.data.name}`, error:err}) + if (consumer.sid) this.removeConsumer(consumer.sid) else { consumer.removeAllListeners() consumer.stream.removeAllListeners() @@ -335,9 +328,9 @@ export default function socketClass(Server) { // TODO do more handling than just logging }) - consumer.authenticated = true + // consumer.authenticated = true - let [err] = await btc(this.authenticateClient)(consumer) + let [err] = await btc(this.authenticateConsumer)(consumer) if (!this.allowAnonymous) { if (err) { consumer.removeAllListeners() @@ -347,18 +340,17 @@ export default function socketClass(Server) { return } } - // authenticated consumer, add to list of clients - consumer.sid = ++this.nextClientID // server assigned ID - consumer.socketSide = true - consumer.authenticated = true - this.clients.push(consumer) // add current consumer to clients + // authenticated consumer, add to list of consumers + consumer.sid = ++this.nextConsumerID // server assigned ID + // consumer.authenticated = true + this.consumers.set(consumer.sid, consumer) // add current consumer to consumers consumer.setKeepAlive(this.keepAlive,30) - const clientCloseHandler = (sid) => { + const consumerCloseHandler = (sid) => { log.warn({msg:'consumer connection was closed',sid:sid}) - this.removeClient(sid) + this.removeConsumer(sid) } - consumer.on('close', clientCloseHandler.bind(this,consumer.sid)) - log.debug({method:'_listen', line:364, msg:'new consumer connected/authenticated', cname:consumer.name, cid:consumer.id, totalConsumers:this.clients.length}) + consumer.on('close', consumerCloseHandler.bind(this,consumer.sid)) + log.debug({method:'_listen', line:364, msg:'new consumer connected/authenticated', cname:consumer.name, cid:consumer.id, totalConsumers:this.consumers.size}) // all's set enable main incoming message processor stream.on('message', messageProcess.bind(this, consumer)) @@ -369,12 +361,17 @@ export default function socketClass(Server) { } this.emit('log',{level:'info', msg:'a consumer connected and authenticated', name:consumer.name, id:consumer.id}) - this.emit('connection:consumer',{state:'connected', msg:'a consumer connected and authenticated', name:consumer.name, id:consumer.id}) + this.emit('connection:consumer',{state:'connected', msg:`consumer ${(consumer.data ||{}).name} connected and authenticated to socket ${this.id}`, + name:(consumer.data ||{}).name ||(consumer.data ||{}).id || consumer.sid, + sid:consumer.sid, + data:consumer.data, + authenticated:consumer.authenticated + }) // that's it. Connection is active - async function messageProcess(client, packet) { - log.debug({method:'_listen', line:179, packet: packet, client:client.name, msg:'incoming packet on socket side'}) + async function messageProcess(consumer, packet) { + log.debug({method:'_listen', line:179, packet: packet, consumer:consumer.data, msg:'incoming packet on socket side'}) let res = (await this._packetProcess(clone(packet))) || {} if (Object.keys(res).length === 0) res = { @@ -390,7 +387,7 @@ export default function socketClass(Server) { res._header.responder = { name: this.name, instanceID: this.id } res._header.socket = this.address() if (!res.cmd) res.cmd = this.defaultReturnCmd || 'reply' // by default return command is 'reply' - let [err] = await btc(this._send)(client,res) + let [err] = await btc(this._send)(consumer,res) if (err) log.error({msg:err, error:err}) } // end message process @@ -402,9 +399,9 @@ export default function socketClass(Server) { // this.push() clearInterval(this._ping) await this.close() - this.clients.forEach(client => { - client.removeAllListeners() - client.stream.removeAllListeners() + this.consumers.forEach(consumer => { + consumer.removeAllListeners() + consumer.stream.removeAllListeners() }) log.debug({method:'_destroy', line:219, msg:'all connections closed....exiting'}) process.exit() @@ -417,18 +414,18 @@ export default function socketClass(Server) { }) } - async _send(client, packet) { - log.trace({msg:`sending to client:${client.id}`, packet:packet}) + async _send(consumer, packet) { + log.trace({msg:`sending to consumer:${consumer.sid}:${consumer.data.name}`, consumer:consumer.data, packet:packet}) return new Promise(async (resolve, reject) => { - if (!client.writable) { + if (!consumer.writable) { reject('socket stream closed can not send packet') return } - let [err,ser] = await btc(client.stream.serialize)(packet) + let [err,ser] = await btc(consumer.stream.serialize)(packet) if (err) reject('unable to serialze the packet') const cb = () => resolve('packet written to socket stream') - if (!client.write(ser)) { - client.once('drain', cb) + if (!consumer.write(ser)) { + consumer.once('drain', cb) } else { process.nextTick(cb) }