176 lines
5.9 KiB
JavaScript
176 lines
5.9 KiB
JavaScript
import { Socket } from 'net'
|
|
import path from 'path'
|
|
import btc from 'better-try-catch'
|
|
import JsonStream from './json-stream'
|
|
|
|
// import logger from '../../uci-logger/src/logger'
|
|
import logger from '@uci/logger'
|
|
|
|
let log = {}
|
|
|
|
// TODO change default pipe dir for windows and mac os
|
|
const DEFAULT_PIPE_DIR = (process.env.SOCKETS_DIR || '/tmp/UCI')
|
|
const DEFAULT_SOCKET_NAME = 'uci-sock'
|
|
|
|
export default class Consumer extends Socket {
|
|
constructor (opts={}) {
|
|
super()
|
|
log = logger({file:'src/consumer.js',class:'Consumer',name:'socket',id:this.id})
|
|
this.id = opts.id || opts.name || 'socket:'+ new Date().getTime()
|
|
if (!opts.path) {
|
|
log.warn({opts:opts},'no host supplied using localhost...use named piped instead')
|
|
opts.host = opts.host || '127.0.0.1'
|
|
opts.port = opts.port || 8080
|
|
} else {
|
|
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.opts=opts
|
|
this.keepAlive = opts.keepAlive ? opts.keepAlive : true
|
|
this._ready = false
|
|
this.timeout = opts.timeout || 500
|
|
this.wait = opts.wait || 5
|
|
this.stream = new JsonStream()
|
|
// bind to class for other class functions
|
|
this.connect = this.connect.bind(this)
|
|
this.__ready = this.__ready.bind(this)
|
|
// this._write = this._write.bind(this)
|
|
}
|
|
|
|
async connect () {
|
|
|
|
return new Promise( (resolve,reject) => {
|
|
|
|
const connect = () => {
|
|
if (this.opts.host ==='127.0.0.1') log.warn('tcp consumer on same machine as host, use named Pipe(Unix) Socket Instead')
|
|
log.info({opts:this.opts},`attempting to connect ${this.id} to socket`)
|
|
super.connect(this.opts)
|
|
}
|
|
|
|
const timeout = setTimeout(() =>{
|
|
reject({opts:this.opts},`unable to connect in ${this.timeout*10}ms`)
|
|
}
|
|
,this.timeout*10)
|
|
|
|
this.once('connect', async () => {
|
|
clearTimeout(timeout)
|
|
this._listen()
|
|
log.info({opts:this.opts},'connected waiting for socket ready handshake')
|
|
this.setKeepAlive(this.keepAlive,100)
|
|
let [err, res] = await btc(isReady).bind(this)(this.__ready, this.wait, this.timeout)
|
|
if (err) reject(err)
|
|
log.info('handshake done, authenticating')
|
|
// TODO authenticate here by encrypting a payload with private key and sending that.
|
|
// await btc(authenticate)
|
|
resolve(res)
|
|
})
|
|
|
|
this.on('error', async (err) => {
|
|
if (err.code === 'EISCONN') {
|
|
return resolve('ready')
|
|
}
|
|
log.warn(err.code)
|
|
setTimeout(() =>{
|
|
// log.warn(`retrying connect to ${this.opts}`)
|
|
connect()
|
|
}
|
|
,this.wait*100)
|
|
})
|
|
|
|
connect()
|
|
|
|
}) //end promise
|
|
|
|
}
|
|
|
|
|
|
async send(ipacket) {
|
|
return new Promise( async (resolve) => {
|
|
// need this for when multiple sends for different consumers use same packet instance
|
|
let packet = Object.assign({},ipacket)
|
|
setTimeout(() => {resolve({error:'no response from socket in 10sec'})},10000)
|
|
packet._header =
|
|
{ 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 },
|
|
path: this.opts.path,
|
|
port: this.opts.port,
|
|
host: this.opts.host
|
|
}
|
|
let [err, res] = await btc(this.stream.serialize)(packet)
|
|
if (err) resolve({error:'unable to serialize packet for sending', packet:packet})
|
|
log.info(await this.__write(res))
|
|
// console.log('listerner set', packet._header.id, packet._header.sender.instanceID)
|
|
this.once(packet._header.id,async function(reply){
|
|
// console.log('reply emitted',reply)
|
|
let res = await this._packetProcess(reply)
|
|
if (!res) { // if process was not promise returning like just logged to console
|
|
res = reply
|
|
log.warn('consumer function was not promise returning - resolving unprocessed')
|
|
}
|
|
resolve(res)
|
|
}) //end listener
|
|
})
|
|
}
|
|
|
|
// TODO register alt stream processor (emit 'message' with JSON, serialize function, onData method for raw socket chucks)
|
|
// TODO register authenciation function (set up default)
|
|
|
|
registerPacketProcessor (func) {
|
|
this._packetProcess = func
|
|
}
|
|
|
|
// PRIVATE METHODS
|
|
|
|
async __write(packet) {
|
|
// timeout already set if sockect can't be drained in 10 secs
|
|
return new Promise(resolve => {
|
|
const cb = () => resolve('=====>packet written to consumer side socket stream ')
|
|
if (!super.write(packet)) {
|
|
this.once('drain',cb )
|
|
} else {
|
|
process.nextTick(cb)
|
|
}
|
|
|
|
})
|
|
}
|
|
|
|
__ready() {return this._ready}
|
|
|
|
async _listen () {
|
|
log.info('listening for incoming packets from socket')
|
|
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) {
|
|
this._ready = true
|
|
return }
|
|
// TODO send back ack with consumer ID and authorization and wait
|
|
// when authorized drop through here to emit
|
|
this.emit(packet._header.id, packet)
|
|
}
|
|
}
|
|
|
|
// default packet process just a simple console logger. ignores any cmd: prop
|
|
_packetProcess (packet) {
|
|
console.log('default consumer processor -- log packet from socket to console')
|
|
console.dir(packet)
|
|
}
|
|
|
|
} // end class
|
|
|
|
// HELP FUNCTIONS
|
|
// wait until a passed ready function returns true
|
|
function isReady(ready, wait=30, timeout=1000) {
|
|
let time = 0
|
|
return new Promise((resolve, reject) => {
|
|
(function waitReady(){
|
|
if (time > timeout) return reject(`timeout waiting for socket ready handshake - ${timeout}ms`)
|
|
if (ready()) return resolve('ready')
|
|
log.info(`waiting ${wait}ms for handshake`)
|
|
time += wait
|
|
setTimeout(waitReady, wait)
|
|
})()
|
|
})
|
|
}
|