2018-01-08 13:06:01 -08:00
|
|
|
import { Socket } from 'net'
|
2018-03-02 08:34:10 -08:00
|
|
|
import path from 'path'
|
2019-03-17 13:51:36 -07:00
|
|
|
import { promisify } from 'util'
|
2018-01-08 13:06:01 -08:00
|
|
|
import btc from 'better-try-catch'
|
2018-02-03 13:33:25 -08:00
|
|
|
import JsonStream from './json-stream'
|
|
|
|
|
2018-02-17 18:26:17 -08:00
|
|
|
// import logger from '../../uci-logger/src/logger'
|
2019-02-14 14:01:14 -08:00
|
|
|
import logger from '@uci-utils/logger'
|
2018-02-17 18:26:17 -08:00
|
|
|
|
2018-02-01 16:58:17 -08:00
|
|
|
let log = {}
|
|
|
|
|
2018-03-02 08:34:10 -08:00
|
|
|
// TODO change default pipe dir for windows and mac os
|
2019-01-01 16:53:12 -08:00
|
|
|
const DEFAULT_PIPE_DIR = process.env.SOCKETS_DIR || '/tmp/UCI'
|
2018-03-02 08:34:10 -08:00
|
|
|
const DEFAULT_SOCKET_NAME = 'uci-sock'
|
2018-01-08 13:06:01 -08:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
/**
|
|
|
|
* Socket Consumer - connects to UCI TCP or Named Pipe Sockets and coummunicates with uci packet.<br/>
|
|
|
|
* Extends {@link https://nodejs.org/api/net.html#net_class_net_socket | nodejs net.Socket}
|
|
|
|
* @extends Socket
|
|
|
|
*/
|
|
|
|
class SocketConsumer extends Socket {
|
|
|
|
/**
|
|
|
|
* constructor - Description
|
|
|
|
*
|
|
|
|
* @param {object} [opts={}] test
|
|
|
|
*/
|
|
|
|
|
|
|
|
constructor(opts = {}) {
|
2018-01-08 13:06:01 -08:00
|
|
|
super()
|
2019-01-01 16:53:12 -08:00
|
|
|
log = logger({
|
|
|
|
file: 'src/consumer.js',
|
|
|
|
class: 'Consumer',
|
|
|
|
name: 'socket',
|
|
|
|
id: this.id
|
|
|
|
})
|
|
|
|
this.id = opts.id || opts.name || 'socket:' + new Date().getTime()
|
2018-02-03 13:33:25 -08:00
|
|
|
if (!opts.path) {
|
2019-02-14 14:01:14 -08:00
|
|
|
if(!opts.host) log.warn({ opts: opts, msg:'no host supplied using localhost...use named piped instead - opts.path'})
|
2018-02-03 13:33:25 -08:00
|
|
|
opts.host = opts.host || '127.0.0.1'
|
|
|
|
opts.port = opts.port || 8080
|
2018-03-02 08:34:10 -08:00
|
|
|
} else {
|
2019-01-01 16:53:12 -08:00
|
|
|
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)
|
2018-03-02 08:34:10 -08:00
|
|
|
}
|
2019-01-01 16:53:12 -08:00
|
|
|
this.opts = opts
|
2018-07-30 19:07:03 -07:00
|
|
|
this.keepAlive = 'keepAlive' in opts ? opts.keepAlive : true
|
2018-01-10 15:03:32 -08:00
|
|
|
this._ready = false
|
2019-03-17 13:51:36 -07:00
|
|
|
this.timeout = opts.timeout || 30 // 5 minutes and then rejects
|
2018-07-31 09:45:32 -07:00
|
|
|
this.wait = opts.wait || 1
|
2018-01-22 12:18:34 -08:00
|
|
|
this.stream = new JsonStream()
|
2018-01-19 20:43:16 -08:00
|
|
|
// bind to class for other class functions
|
|
|
|
this.connect = this.connect.bind(this)
|
2019-03-17 13:51:36 -07:00
|
|
|
this.close = promisify(this.end).bind(this)
|
2018-02-01 16:58:17 -08:00
|
|
|
this.__ready = this.__ready.bind(this)
|
2018-02-12 14:41:06 -08:00
|
|
|
// this._write = this._write.bind(this)
|
2018-01-08 13:06:01 -08:00
|
|
|
}
|
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
async connect() {
|
|
|
|
return new Promise((resolve, reject) => {
|
2018-01-30 16:59:57 -08:00
|
|
|
const connect = () => {
|
2019-01-01 16:53:12 -08:00
|
|
|
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`
|
|
|
|
)
|
2018-02-03 13:33:25 -08:00
|
|
|
super.connect(this.opts)
|
2018-01-30 16:59:57 -08:00
|
|
|
}
|
|
|
|
|
2018-07-31 09:45:32 -07:00
|
|
|
let reconnect = {}
|
2019-03-17 13:51:36 -07:00
|
|
|
// this is only for initial connection
|
2019-01-01 16:53:12 -08:00
|
|
|
const timeout = setTimeout(() => {
|
2018-07-31 09:45:32 -07:00
|
|
|
clearTimeout(reconnect)
|
2019-01-01 16:53:12 -08:00
|
|
|
log.fatal({ opts: this.opts }, `unable to connect in ${this.timeout}s`)
|
|
|
|
reject(
|
|
|
|
{ opts: this.opts },
|
|
|
|
`unable to connect to socket server in ${this.timeout}secs`
|
|
|
|
)
|
|
|
|
}, this.timeout * 1000)
|
2018-01-08 13:06:01 -08:00
|
|
|
|
2018-01-30 16:59:57 -08:00
|
|
|
this.once('connect', async () => {
|
|
|
|
clearTimeout(timeout)
|
2018-01-31 11:30:32 -08:00
|
|
|
this._listen()
|
2019-01-01 16:53:12 -08:00
|
|
|
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
|
|
|
|
)
|
2018-01-13 20:46:14 -08:00
|
|
|
if (err) reject(err)
|
2018-02-01 16:58:17 -08:00
|
|
|
log.info('handshake done, authenticating')
|
2018-01-30 16:59:57 -08:00
|
|
|
// TODO authenticate here by encrypting a payload with private key and sending that.
|
|
|
|
// await btc(authenticate)
|
2018-01-13 20:46:14 -08:00
|
|
|
resolve(res)
|
|
|
|
})
|
2018-01-20 13:52:07 -08:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
this.on('error', async err => {
|
|
|
|
log.warn({ error: err.code }, `connect error ${err.code}`)
|
2018-01-30 16:59:57 -08:00
|
|
|
if (err.code === 'EISCONN') {
|
|
|
|
return resolve('ready')
|
|
|
|
}
|
2018-07-30 19:07:03 -07:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
reconnect = setTimeout(() => {
|
|
|
|
connect()
|
|
|
|
}, this.wait * 1000)
|
2018-07-30 19:07:03 -07:00
|
|
|
})
|
|
|
|
|
|
|
|
this.on('end', async () => {
|
|
|
|
log.warn('socket (server) terminated unexpectantly')
|
|
|
|
if (this.keepAlive) {
|
2019-01-01 16:53:12 -08:00
|
|
|
log.info(
|
|
|
|
'keep alive was set, so waiting on server to come online for reconnect'
|
|
|
|
)
|
2018-07-30 19:07:03 -07:00
|
|
|
this.destroy()
|
2019-01-01 16:53:12 -08:00
|
|
|
this.emit('error', { code: 'DISCONNECTED' })
|
2018-01-30 16:59:57 -08:00
|
|
|
}
|
|
|
|
})
|
2018-01-10 15:03:32 -08:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
connect() // initial connect request
|
2018-01-30 16:59:57 -08:00
|
|
|
}) //end promise
|
2018-01-08 13:06:01 -08:00
|
|
|
}
|
|
|
|
|
2018-02-13 13:51:58 -08:00
|
|
|
async send(ipacket) {
|
2019-01-01 16:53:12 -08:00
|
|
|
return new Promise(async resolve => {
|
2018-02-13 13:51:58 -08:00
|
|
|
// need this for when multiple sends for different consumers use same packet instance
|
2019-01-01 16:53:12 -08:00
|
|
|
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 },
|
2018-02-13 13:51:58 -08:00
|
|
|
path: this.opts.path,
|
|
|
|
port: this.opts.port,
|
|
|
|
host: this.opts.host
|
|
|
|
}
|
2018-02-12 14:41:06 -08:00
|
|
|
let [err, res] = await btc(this.stream.serialize)(packet)
|
2019-01-01 16:53:12 -08:00
|
|
|
if (err)
|
|
|
|
resolve({
|
|
|
|
error: 'unable to serialize packet for sending',
|
|
|
|
packet: packet
|
|
|
|
})
|
2018-07-31 09:45:32 -07:00
|
|
|
await this.__write(res)
|
2019-01-01 16:53:12 -08:00
|
|
|
this.once(packet._header.id, async function(reply) {
|
2018-02-12 14:41:06 -08:00
|
|
|
let res = await this._packetProcess(reply)
|
2019-01-01 16:53:12 -08:00
|
|
|
if (!res) {
|
|
|
|
// if process was not promise returning like just logged to console
|
2018-02-12 14:41:06 -08:00
|
|
|
res = reply
|
2018-07-31 09:45:32 -07:00
|
|
|
// log.warn('consumer function was not promise returning further processing may be out of sequence')
|
2018-02-13 13:51:58 -08:00
|
|
|
}
|
2018-02-12 14:41:06 -08:00
|
|
|
resolve(res)
|
|
|
|
}) //end listener
|
|
|
|
})
|
2018-01-25 18:07:45 -08:00
|
|
|
}
|
|
|
|
|
2018-01-30 16:59:57 -08:00
|
|
|
// TODO register alt stream processor (emit 'message' with JSON, serialize function, onData method for raw socket chucks)
|
|
|
|
// TODO register authenciation function (set up default)
|
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
registerPacketProcessor(func) {
|
2018-02-03 13:33:25 -08:00
|
|
|
this._packetProcess = func
|
2018-01-25 18:07:45 -08:00
|
|
|
}
|
|
|
|
|
2018-02-03 13:33:25 -08:00
|
|
|
// PRIVATE METHODS
|
|
|
|
|
2018-02-12 14:41:06 -08:00
|
|
|
async __write(packet) {
|
|
|
|
// timeout already set if sockect can't be drained in 10 secs
|
|
|
|
return new Promise(resolve => {
|
2018-07-31 09:45:32 -07:00
|
|
|
const cb = () => resolve('packet written to consumer side socket stream ')
|
2018-02-12 14:41:06 -08:00
|
|
|
if (!super.write(packet)) {
|
2019-01-01 16:53:12 -08:00
|
|
|
this.once('drain', cb)
|
2018-02-12 14:41:06 -08:00
|
|
|
} else {
|
|
|
|
process.nextTick(cb)
|
|
|
|
}
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
__ready() {
|
|
|
|
return this._ready
|
|
|
|
}
|
2018-02-01 16:58:17 -08:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
async _listen() {
|
2018-02-01 16:58:17 -08:00
|
|
|
log.info('listening for incoming packets from socket')
|
2018-05-24 12:28:30 -07:00
|
|
|
// listen for pushed packets
|
2019-01-01 16:53:12 -08:00
|
|
|
this.on('pushed', async function(packet) {
|
2018-05-24 12:28:30 -07:00
|
|
|
// TODO do some extra security here?
|
|
|
|
let res = await this._packetProcess(packet)
|
2019-01-01 16:53:12 -08:00
|
|
|
if (!res) {
|
|
|
|
// if process was not promise returning like just logged to console
|
2019-03-17 13:51:36 -07:00
|
|
|
log.warn('consumer packet processing function was not promise returning')
|
2018-05-24 12:28:30 -07:00
|
|
|
}
|
|
|
|
})
|
|
|
|
// listen on socket stream
|
2018-02-01 16:58:17 -08:00
|
|
|
this.on('data', this.stream.onData)
|
|
|
|
this.stream.on('message', messageProcess.bind(this))
|
2018-05-24 12:28:30 -07:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
async function messageProcess(packet) {
|
2018-02-13 13:51:58 -08:00
|
|
|
// console.log('incoming packet from socket',packet)
|
2018-02-11 19:58:22 -08:00
|
|
|
if (packet._handshake) {
|
2018-02-01 16:58:17 -08:00
|
|
|
this._ready = true
|
2019-01-01 16:53:12 -08:00
|
|
|
return
|
|
|
|
}
|
2018-02-12 14:41:06 -08:00
|
|
|
// TODO send back ack with consumer ID and authorization and wait
|
|
|
|
// when authorized drop through here to emit
|
2018-02-13 13:51:58 -08:00
|
|
|
this.emit(packet._header.id, packet)
|
2018-02-01 16:58:17 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-02-12 14:41:06 -08:00
|
|
|
// default packet process just a simple console logger. ignores any cmd: prop
|
2019-01-01 16:53:12 -08:00
|
|
|
_packetProcess(packet) {
|
|
|
|
console.log(
|
|
|
|
'default consumer processor -- log packet from socket to console'
|
|
|
|
)
|
2018-02-03 13:33:25 -08:00
|
|
|
console.dir(packet)
|
|
|
|
}
|
2019-01-01 16:53:12 -08:00
|
|
|
} // end class
|
2018-02-03 13:33:25 -08:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
export default SocketConsumer
|
2018-01-10 15:03:32 -08:00
|
|
|
|
2019-01-01 16:53:12 -08:00
|
|
|
// Helper Functions
|
2018-02-03 13:33:25 -08:00
|
|
|
// wait until a passed ready function returns true
|
2019-01-01 16:53:12 -08:00
|
|
|
function isReady(ready, wait = 30, timeout = 1000) {
|
2018-01-10 15:03:32 -08:00
|
|
|
let time = 0
|
2018-01-13 20:46:14 -08:00
|
|
|
return new Promise((resolve, reject) => {
|
2019-01-01 16:53:12 -08:00
|
|
|
(function waitReady() {
|
|
|
|
if (time > timeout)
|
|
|
|
return reject(
|
|
|
|
`timeout waiting for socket ready handshake - ${timeout}ms`
|
|
|
|
)
|
2018-01-10 15:03:32 -08:00
|
|
|
if (ready()) return resolve('ready')
|
2018-01-21 19:55:47 -08:00
|
|
|
log.info(`waiting ${wait}ms for handshake`)
|
|
|
|
time += wait
|
2018-01-13 20:46:14 -08:00
|
|
|
setTimeout(waitReady, wait)
|
2018-01-10 15:03:32 -08:00
|
|
|
})()
|
|
|
|
})
|
|
|
|
}
|