Compare commits
7 Commits
http-serve
...
master
Author | SHA1 | Date |
---|---|---|
David Kebler | 9eeb5e29a2 | |
David Kebler | aeae1df3fa | |
David Kebler | 84cd362751 | |
David Kebler | 87e94b765b | |
David Kebler | 87fba5c955 | |
David Kebler | 4fcdc4166a | |
David Kebler | 88c61f0895 |
|
@ -1,30 +0,0 @@
|
|||
import Socket from '../src'
|
||||
|
||||
async function packetProcess(packet) {
|
||||
// Here one registers a packet processor which gets an plain object hash from the client
|
||||
// should return a promise
|
||||
// for example if client sends a property cmd on can take action on that
|
||||
return new Promise(resolve => {
|
||||
let res = {}
|
||||
switch (packet.cmd) {
|
||||
case 'echo':
|
||||
res.msg = 'Echoing Back any payload propery'
|
||||
res.payload = packet.payload
|
||||
this.push(res)
|
||||
break
|
||||
default:
|
||||
res.msg = `.cmd, command ${packet.cmd} was unknown at server`
|
||||
res.sent_packet = packet
|
||||
}
|
||||
resolve(res)
|
||||
})
|
||||
}
|
||||
|
||||
// let test = new Test()
|
||||
let test = new Socket({ port: 8090, clientTracking: true })
|
||||
test.registerPacketProcessor(packetProcess)
|
||||
;(async () => {
|
||||
console.log(await test.create())
|
||||
})().catch(err => {
|
||||
console.error('FATAL: UNABLE TO START SYSTEM!\n', err)
|
||||
})
|
16
package.json
16
package.json
|
@ -1,11 +1,12 @@
|
|||
{
|
||||
"name": "@uci/websocket",
|
||||
"version": "0.3.8",
|
||||
"version": "0.4.1",
|
||||
"description": "JSON packet host websocket server",
|
||||
"main": "src",
|
||||
"scripts": {
|
||||
"example": "node -r esm examples/server",
|
||||
"dev": "UCI_ENV=dev UCI_LOG_LEVEL=debug ./node_modules/.bin/nodemon -r esm examples/server"
|
||||
"server": "node -r esm examples/server",
|
||||
"server:dev": "UCI_ENV=dev ./node_modules/.bin/nodemon -r esm examples/server",
|
||||
"server:debug": "UCI_LOG_LEVEL=debug npm run server:dev"
|
||||
},
|
||||
"author": "David Kebler",
|
||||
"license": "MIT",
|
||||
|
@ -32,15 +33,16 @@
|
|||
"chai": "^4.2.0",
|
||||
"chai-as-promised": "^7.1.1",
|
||||
"esm": "^3.2.25",
|
||||
"mocha": "^6.2.0",
|
||||
"nodemon": "^1.19.1"
|
||||
"mocha": "^8.0.1",
|
||||
"nodemon": "^2.0.4"
|
||||
},
|
||||
"dependencies": {
|
||||
"@uci-utils/logger": "0.0.15",
|
||||
"@uci-utils/logger": "0.0.18",
|
||||
"await-to-js": "^2.1.1",
|
||||
"better-try-catch": "^0.6.2",
|
||||
"clone": "^2.1.2",
|
||||
"death": "^1.1.0",
|
||||
"ws": "^7.1.2"
|
||||
"p-reflect": "^2.1.0",
|
||||
"ws": "^7.3.1"
|
||||
}
|
||||
}
|
||||
|
|
373
src/socket.js
373
src/socket.js
|
@ -1,6 +1,7 @@
|
|||
import WebSocket from 'ws'
|
||||
import { Server as WSS } from 'ws'
|
||||
import { Server } from 'http'
|
||||
import btc from 'better-try-catch'
|
||||
import to from 'await-to-js'
|
||||
import pReflect from 'p-reflect'
|
||||
import { promisify } from 'util'
|
||||
import _ON_DEATH from 'death' //this is intentionally ugly
|
||||
import clone from 'clone'
|
||||
|
@ -9,24 +10,41 @@ import logger from '@uci-utils/logger'
|
|||
let log = {}
|
||||
|
||||
/**
|
||||
* Socket - a websocket that supports uci packets
|
||||
* @class WebSocket
|
||||
* @extends ws.server
|
||||
*
|
||||
* Socket - Description
|
||||
* @extends Server
|
||||
*/
|
||||
class Socket extends WebSocket.Server {
|
||||
class Socket extends Server {
|
||||
|
||||
/**
|
||||
* constructor - Description
|
||||
*
|
||||
* @param {object} [opts={}] Description
|
||||
*
|
||||
* @returns {type} Description
|
||||
*/
|
||||
constructor(opts = {}) {
|
||||
super(opts)
|
||||
opts.host = opts.host || '0.0.0.0'
|
||||
opts.port = opts.port || 8090
|
||||
opts.clientTracking = opts.clientTracking || true
|
||||
super(opts)
|
||||
this.id = opts.id || opts.name || 'Websocket:' + new Date().getTime()
|
||||
this.opts = opts // for use to recover from selected errors
|
||||
//self bindings
|
||||
this._listen = this._listen.bind(this)
|
||||
this.wss = {} // web socket instance goes here
|
||||
this.allowAnonymous = (!opts.tokens || !!process.env.UCI_ANON || opts.allowAnonymous) ? true : false
|
||||
this.tokens = opts.tokens || []
|
||||
this.pingInterval = opts.pingInterval === false ? opts.pingInterval : (opts.pingInterval * 1000 || 5000)
|
||||
this.nextconsumerID = 0 // incrementer for default initial consumer ID
|
||||
this.consumers = new Map()
|
||||
this.conPackets = opts.conPackets || [opts.conPacket]
|
||||
this.errors = []
|
||||
this.errorCount = 0
|
||||
this.create = this.create.bind(this)
|
||||
this._destroy = this._destroy.bind(this)
|
||||
this._listening=false
|
||||
this.authenticateconsumer = this.authenticateconsumer.bind(this)
|
||||
this._authenticate = this._authenticate.bind(this)
|
||||
this.close = promisify(this.close).bind(this)
|
||||
log = logger({
|
||||
package:'@uci/websocket',
|
||||
file: 'src/socket.js',
|
||||
class: 'Socket',
|
||||
name: 'websocket',
|
||||
|
@ -34,76 +52,273 @@ class Socket extends WebSocket.Server {
|
|||
})
|
||||
} // end constructor
|
||||
|
||||
get active() {
|
||||
return this._listening
|
||||
}
|
||||
|
||||
/**
|
||||
* create - Description
|
||||
*
|
||||
* @returns {type} Description
|
||||
*/
|
||||
async create() {
|
||||
this.emit('socket',{state:'creating', msg:'creating socket for consumers to connect'})
|
||||
return new Promise((resolve, reject) => {
|
||||
_ON_DEATH(async () => {
|
||||
log.error({method:'create', line:39, msg:'\nhe\'s dead jim'})
|
||||
log.error({method:'create', line:51, msg:'\nhe\'s dead jim'})
|
||||
await this._destroy()
|
||||
})
|
||||
|
||||
// kills nodemon properly
|
||||
process.once('SIGUSR2', async () => {
|
||||
await this._destroy
|
||||
await this._destroy()
|
||||
process.kill(process.pid, 'SIGUSR2')
|
||||
})
|
||||
|
||||
this.on('error', async err => {
|
||||
log.error({method:'creaete', line:48, err:err, msg:'socket server error'})
|
||||
this.once('error', async err => {
|
||||
log.error({method:'creaete', line:48, err:err, msg:'http server error'})
|
||||
reject(err)
|
||||
})
|
||||
|
||||
this.on('listening', async () => {
|
||||
this._listen()
|
||||
this.once('listening', () => {
|
||||
log.info({method:'create', line:54, msg:'websocket server created and listening at', address:this.address()})
|
||||
resolve(
|
||||
`websocket ready and listening at ${this.address().address}:${
|
||||
this.address().port
|
||||
}`
|
||||
)
|
||||
// emit ready
|
||||
this.on('error', err => {
|
||||
this.errorCount +=1 // log errors here
|
||||
this.errors.push(err)
|
||||
if(this.errorCount>2) {
|
||||
this.emit('log', {level:'warn',msg:'something bad maybe going on, 3 errors', errors:this.errors})
|
||||
this.emit('socket',{state:'error', msg:'2 to 5 socket errors', errors:this.errors})
|
||||
}
|
||||
if(this.errorCount>5) {
|
||||
let errors = {level:'fatal',msg:'something fatal is going on, 6 errors', errors:this.errors}
|
||||
log.fatal(errors)
|
||||
this._listening=false
|
||||
this.close(() => {
|
||||
this.emit('socket',{state:'offline', msg:'too many socket errors no longer listening for consumers to connect'})
|
||||
})
|
||||
this.emit('log', {active:this.active})
|
||||
this.emit('log', errors)
|
||||
}
|
||||
})
|
||||
this.wss = new WSS({server:this})
|
||||
this.wss.on('error', err => {this.emit('error', err)}) // bubble up errors
|
||||
this.wss.on('connection', this._connectionHandler.bind(this))
|
||||
this._listening=true
|
||||
let msg = `socket ready and listening ${typeof this.address() ==='string' ? `at ${this.address()}` : `on port ${this.address().port}`}`
|
||||
this.emit('log',{active:this.active})
|
||||
this.emit('socket',{state:'listening', msg:msg})
|
||||
resolve(`websocket ready and listening at ${this.address().address}:${this.address().port}`)
|
||||
})
|
||||
super.listen(this.opts)
|
||||
})
|
||||
} // end create
|
||||
|
||||
registerPacketProcessor(func) {
|
||||
this._packetProcess = func
|
||||
this._packetProcess = func.bind(this)
|
||||
}
|
||||
|
||||
_listen() {
|
||||
this.on('connection', async (socket, req) => {
|
||||
let send = this._send.bind(socket)
|
||||
log.debug({method:'_listen', line:71, req: req, msg: 'new consumer connecting'})
|
||||
socket.address = req.remoteAddress
|
||||
addTokens(tokens) {
|
||||
if (typeof tokens ==='string'){
|
||||
tokens = tokens.split(',')
|
||||
}
|
||||
this.tokens = this.tokens.concat(tokens)
|
||||
if (this.tokens.length>0) this.allowAnonymous = false
|
||||
}
|
||||
|
||||
socket.on('error', (err) => {
|
||||
log.error({msg:'client connection error during listen',error:err})
|
||||
})
|
||||
removeTokens(tokens) {
|
||||
if (typeof tokens ==='string'){
|
||||
if (tokens === 'all') {
|
||||
this.tokens = []
|
||||
this.allowAnonymous = true
|
||||
return
|
||||
}
|
||||
tokens = tokens.split(',')
|
||||
}
|
||||
this.tokens = this.tokens.filter(token => !tokens.includes(token))
|
||||
if (this.tokens.length===0) {
|
||||
log.warn({msg:'all tokens have been removed, switching to allow anonymous connections'})
|
||||
this.allowAnonymous = true
|
||||
}
|
||||
}
|
||||
|
||||
socket.on('close', (msg) => {
|
||||
log.warn({msg:'client connection closed during listen',error:msg})
|
||||
})
|
||||
registerTokenValidator (func) {
|
||||
this.allowAnonymous = false
|
||||
this._validateToken = func
|
||||
}
|
||||
|
||||
registerAuthenticator (func) {
|
||||
this.allowAnonymous = false
|
||||
this._authenticate = func
|
||||
}
|
||||
|
||||
|
||||
socket.on('message', messageProcess.bind(this, socket))
|
||||
|
||||
async function messageProcess(client, strPacket) {
|
||||
log.debug({method:'_listen', line:76, packet: strPacket, msg:' incoming packet on web socket side'})
|
||||
let res = {}
|
||||
let [err, packet] = btc(JSON.parse)(strPacket)
|
||||
log.debug({method:'_listen', line:79, error:err, packet:packet, msg:'parsed websocket packet'})
|
||||
if (err) {
|
||||
res = { error: `Could not parse JSON: ${packet}` }
|
||||
} else {
|
||||
if (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
|
||||
}
|
||||
/**
|
||||
* push
|
||||
*
|
||||
* @param {Object} packet - this is the parameter packet
|
||||
* @param {<type>} id - this is the parameter id
|
||||
*
|
||||
*/
|
||||
async push(packet={},opts={}) {
|
||||
if (this.consumers.size > 0) {
|
||||
packet._header = {id: opts.packetId || 'pushed'}
|
||||
let consumers = []
|
||||
if ( opts.consumers || opts.consumer ) {
|
||||
if (opts.consumer) opts.consumers = [opts.consumer]
|
||||
consumers = Array.from(this.consumers).filter(([sid,consumer]) =>
|
||||
opts.consumers.some(con=> {
|
||||
console.log(consumer.sid,consumer.data,con)
|
||||
return (
|
||||
con === ( (consumer.data ||{}).name || (consumer.data ||{}).id ) ||
|
||||
con.sid=== sid ||
|
||||
con.name === (consumer.data ||{}).name ||
|
||||
con.id === (consumer.data ||{}).id
|
||||
)
|
||||
}
|
||||
)
|
||||
).map(con=>con[1])
|
||||
// console.log('custom consumers',consumers.length)
|
||||
} else consumers = Array.from(this.consumers.values())
|
||||
|
||||
consumers = consumers.filter(consumer=>consumer.writable||consumer.readyState===1)
|
||||
|
||||
const send = consumer => this._send(consumer,packet)
|
||||
const res = await Promise.all(consumers.map(send).map(pReflect))
|
||||
|
||||
const success = res.filter(result => result.isFulfilled).map((result,index) => [consumers[index].name,result.value])
|
||||
const errors =res.filter(result => result.isRejected).map((result,index) => [consumers[index].name,result.reason])
|
||||
this.emit('log',{level:errors.length? 'error': packet._header.id==='ping'?'trace':'debug', msg:'packet was pushed', socket:this.name||this.id, errors:errors, packet:packet, success:success, headerId:packet._header.id})
|
||||
} else {
|
||||
this.emit('log',{level:'debug', msg:'no connected consumers packet push ignored', packet:packet})
|
||||
}
|
||||
}
|
||||
|
||||
removeconsumer (id) {
|
||||
let consumer = this.consumers.get(id)
|
||||
this.emit('connection:consumer',{state:'disconnected', name:consumer.name})
|
||||
clearInterval(consumer._ping)
|
||||
consumer.removeAllListeners()
|
||||
log.warn({msg:`consumer ${id}:${consumer.name} removed from server tracking`, id:id, name:consumer.name, curconsumerCount:this.consumers.size})
|
||||
this.consumers.delete(id)
|
||||
|
||||
}
|
||||
|
||||
async authenticateconsumer(consumer) {
|
||||
// let server = this
|
||||
return new Promise(async function(resolve, reject) {
|
||||
// when consumer gets the handshake they must follow with authentication
|
||||
consumer.on('message', authenticate.bind(this,consumer))
|
||||
let [err] = await btc(this._send)(consumer,{_handshake: true, id:consumer.id})
|
||||
if (err) {
|
||||
log.error({msg:'error in handshake send', error:err})
|
||||
reject(err)
|
||||
}
|
||||
|
||||
async function authenticate (consumer,message) {
|
||||
let [err,packet] = btc(JSON.parse)(message)
|
||||
if (err) reject('unable to parse authentication packet from consumer')
|
||||
log.debug({msg:`authentication packet from consumer ${consumer.id}`, packet:packet})
|
||||
if (!packet._authenticate) reject('first consumer packet was not authentication')
|
||||
else {
|
||||
let [err, res] = await btc(this._authenticate)(packet)
|
||||
consumer.authenticated = this.allowAnonymous ? 'anonymous' : (err ? false : res)
|
||||
consumer.name = packet.consumerName
|
||||
packet.authenticated = consumer.authenticated
|
||||
if (err && !this.allowAnonymous) packet.reason = err
|
||||
log.debug({msg:'sending authorization result to consumer', packet:packet})
|
||||
await this._send(consumer,packet) // send either way
|
||||
if (packet.reason) {
|
||||
log.info({msg:'consumer authentication failed', consumer:consumer.name, consumer_id:consumer.id, reason:err})
|
||||
reject(packet.reason)
|
||||
}
|
||||
else {
|
||||
log.info({msg:'consumer authenticated successfuly', consumer:consumer.name, consumer_id:consumer.id})
|
||||
if (this.allowAnonymous) log.warn({msg:'web consumer connected anonymously', consumer:consumer.name, consumer_id:consumer.id})
|
||||
resolve(consumer.authenticated)
|
||||
}
|
||||
}
|
||||
}
|
||||
}.bind(this))
|
||||
}
|
||||
|
||||
// private methods
|
||||
|
||||
|
||||
// default validator
|
||||
_validateToken (token) {
|
||||
if (token) return this.tokens.includes(token)
|
||||
return false
|
||||
}
|
||||
|
||||
// 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) {
|
||||
log.debug({method:'_connectionHandler', line:76, msg: 'new web consumer connecting'})
|
||||
|
||||
consumer.id = ++this.nextconsumerID // server assigned ID
|
||||
this.consumers.set(consumer.id,consumer)
|
||||
consumer.authenticated = false
|
||||
consumer.connected = true
|
||||
|
||||
// add listeners
|
||||
const consumerCloseHandler = (id) => {
|
||||
log.warn({msg:'consumer connection closed during listen,',id:id})
|
||||
this.removeconsumer(id)
|
||||
}
|
||||
|
||||
consumer.on('close', consumerCloseHandler.bind(this,consumer.id))
|
||||
|
||||
consumer.on('error', (err) => {
|
||||
log.error({msg:'consumer connection error during listen',error:err})
|
||||
// TODO do more handling than just logging
|
||||
})
|
||||
|
||||
let [err] = await btc(this.authenticateconsumer)(consumer)
|
||||
if (!this.allowAnonymous) {
|
||||
if (err) {
|
||||
consumer.close()// abort new connection consumer, cleanup, remove listeners
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if (this.conPackets.length) {
|
||||
for (let packet of this.conPackets) {
|
||||
packet._header = {type:'on connection packet', id: 'pushed'}
|
||||
await this._send(consumer,packet) // send a packet command on to consumer on connection
|
||||
}
|
||||
}
|
||||
|
||||
this.emit('connection:consumer',{state:'connected', name:consumer.name, id:consumer.id})
|
||||
|
||||
consumer._ping = setInterval( () => {
|
||||
consumer.ping(JSON.stringify({pingInterval:this.pingInterval}))
|
||||
this._send(consumer,{_header:{id:'ping'},pingInterval:this.pingInterval})
|
||||
},this.pingInterval)
|
||||
|
||||
consumer.on('message', messageProcess.bind(this, consumer))
|
||||
|
||||
async function messageProcess(consumer, strPacket) {
|
||||
log.debug({method:'_listen', line:76, packet: strPacket, msg:' incoming packet from web consumer'})
|
||||
let res = {}
|
||||
let [err, packet] = btc(JSON.parse)(strPacket)
|
||||
if (err) {
|
||||
res = { error: 'Could not JSON parse packet', packet:strPacket }
|
||||
}
|
||||
else {
|
||||
log.debug({method:'_listen', line:266, packet:packet, msg:'parsed packet'})
|
||||
res = (await this._packetProcess(clone(packet))) || {}
|
||||
if (Object.keys(res).length === 0)
|
||||
res = {
|
||||
error:
|
||||
'consumer 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
|
||||
|
@ -112,17 +327,21 @@ class Socket extends WebSocket.Server {
|
|||
res._header.responder = { name: this.name, instanceID: this.id }
|
||||
res._header.socket = this.address()
|
||||
if (!res.cmd) res.cmd = 'reply' // by default return command is 'reply'
|
||||
let sres = await send(res)
|
||||
log.debug({method:'_listen', line:105, packet: res, sendResponse:sres, msg:'packet reply send back to client'})
|
||||
let [err] = await btc(this._send)(consumer,res)
|
||||
if (err) log.error({msg:err, error:err})
|
||||
}
|
||||
}) // end connected consumer
|
||||
log.debug('socket created')
|
||||
} // end listen
|
||||
} // end message process
|
||||
|
||||
} // end connection Handler
|
||||
|
||||
async _destroy() {
|
||||
log.debug({method:'_listen', line:105, msg:'closing down socket'})
|
||||
log.debug({method:'_destroy', line:302, msg:'closing down http server and attached sockets', port:this.port})
|
||||
this.consumers.forEach(consumer => {
|
||||
consumer.terminate()
|
||||
consumer.emit('close')
|
||||
})
|
||||
await this.close()
|
||||
log.debug({method:'_listen', line:105, msg:'all connections closed....exiting'})
|
||||
log.debug({method:'_listen', line:105, msg:'all connections closed, all consumer sockets deleted....exiting'})
|
||||
process.exit()
|
||||
}
|
||||
|
||||
|
@ -133,27 +352,23 @@ class Socket extends WebSocket.Server {
|
|||
})
|
||||
}
|
||||
|
||||
async push(packet, id) {
|
||||
packet._header = { id: id || 'pushed' }
|
||||
this.clients.forEach(async client => {
|
||||
if (client.readyState === WebSocket.OPEN) {
|
||||
if (!id || id === client.ID) await this._send.bind(client)(packet)
|
||||
}
|
||||
|
||||
|
||||
async _send(consumer, packet) {
|
||||
log.debug({msg:`sending to client:${consumer.id}:${consumer.name}`, packet:packet})
|
||||
return new Promise(async (resolve, reject) => {
|
||||
// if (!consumer._connected) reject('can not send no connection')
|
||||
let [err, message] = btc(JSON.stringify)(packet)
|
||||
if (err) reject('unable to serialze the packet')
|
||||
consumer.send(message,(err) => {
|
||||
if (!err) resolve('packet written to socket stream successfully')
|
||||
else reject(`error sending: ${err}`)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// must have a consumer socket instance bound to call this function
|
||||
async _send(packet) {
|
||||
if (this.readyState !== 1) log.error({method:'_send', line:147, msg:`Connection not Ready, CODE:${this.readyState}`})
|
||||
else {
|
||||
let [err, message] = btc(JSON.stringify)(packet)
|
||||
if (err) log.error({method:'_send', line:147, msg:`WS Could not JSON stringify: ${packet}`})
|
||||
else {
|
||||
try { this.send(message) }
|
||||
catch(err) { log.error({method:'_send', line:153, msg:'error sending from ws server', error:err}) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
} // end class
|
||||
|
||||
export default Socket
|
||||
|
|
|
@ -1,2 +0,0 @@
|
|||
*.sock
|
||||
/node_modules/
|
|
@ -1 +0,0 @@
|
|||
/node_modules/
|
|
@ -1,4 +0,0 @@
|
|||
export default async function (packet) {
|
||||
packet.payload = this.opts.port +':'+packet.payload
|
||||
return packet
|
||||
}
|
|
@ -1,14 +0,0 @@
|
|||
import { Socket } from '../../src'
|
||||
import process from './tcp-process'
|
||||
|
||||
let socket = new Socket({port:9080, name:'tcp socket 9080'})
|
||||
|
||||
socket.registerPacketProcessor(process)
|
||||
;
|
||||
(async () => {
|
||||
|
||||
await socket.create()
|
||||
|
||||
})().catch(err => {
|
||||
console.error('FATAL: UNABLE TO START SYSTEM!\n',err)
|
||||
})
|
|
@ -1,15 +0,0 @@
|
|||
import { Socket } from '../../src'
|
||||
import process from './tcp-process'
|
||||
|
||||
let socket = new Socket({name:'tcp socket'})
|
||||
|
||||
socket.registerPacketProcessor(process)
|
||||
|
||||
;
|
||||
(async () => {
|
||||
|
||||
await socket.create()
|
||||
|
||||
})().catch(err => {
|
||||
console.error('FATAL: UNABLE TO START SYSTEM!\n',err)
|
||||
})
|
|
@ -1,11 +0,0 @@
|
|||
import { Socket } from '../../src'
|
||||
|
||||
let socket = new Socket({path:true,name:'default-unix-socket'})
|
||||
|
||||
;
|
||||
(async () => {
|
||||
await socket.create()
|
||||
|
||||
})().catch(err => {
|
||||
console.error('FATAL: UNABLE TO START SYSTEM!\n',err)
|
||||
})
|
|
@ -1,23 +0,0 @@
|
|||
import { Socket } from '../../src'
|
||||
|
||||
const USOCKET = 'usocket'
|
||||
|
||||
let socket = new Socket({path:USOCKET,name:'default-unix-socket'})
|
||||
|
||||
socket.test = 'at socket => '
|
||||
|
||||
socket.registerPacketProcessor(async function (packet) {
|
||||
return new Promise((resolve) => {
|
||||
packet.payload = 'overwrite default processor from instance '+ this.test + packet.payload
|
||||
resolve(packet)
|
||||
})
|
||||
})
|
||||
|
||||
;
|
||||
(async () => {
|
||||
|
||||
await socket.create()
|
||||
|
||||
})().catch(err => {
|
||||
console.error('FATAL: UNABLE TO START SYSTEM!\n',err)
|
||||
})
|
|
@ -1,78 +0,0 @@
|
|||
import { spawn } from 'child_process'
|
||||
import chai from 'chai'
|
||||
import chaiAsPromised from 'chai-as-promised'
|
||||
import btc from 'better-try-catch'
|
||||
chai.use(chaiAsPromised)
|
||||
const expect = chai.expect
|
||||
|
||||
import { Consumer } from '../src'
|
||||
|
||||
let tcpsocket_default = {}
|
||||
let tcpsocket_9080 = {}
|
||||
|
||||
describe('Connects and Processes a payload in a JSON packet via TCP Socket', function(){
|
||||
|
||||
before(async function(){
|
||||
tcpsocket_default = spawn('node',['-r', '@std/esm', './test/sockets/tcpsocket-default'])
|
||||
tcpsocket_default.stdout.on('data', function(buf) {
|
||||
console.log('[Socket]', String(buf))
|
||||
})
|
||||
|
||||
tcpsocket_9080 = spawn('node',['-r', '@std/esm', './test/sockets/tcpsocket-9080'])
|
||||
tcpsocket_9080.stdout.on('data', function(buf) {
|
||||
console.log('[Socket]', String(buf))
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
after(async function(){
|
||||
tcpsocket_default.kill()
|
||||
tcpsocket_9080.kill()
|
||||
})
|
||||
|
||||
it('with default host and port', async function () {
|
||||
let tcpconsumer_default = new Consumer({name:'tcpconsumer'})
|
||||
|
||||
let [err] = await btc(tcpconsumer_default.connect)()
|
||||
if (err) {
|
||||
console.log('unable to connect to socket to start test', tcpconsumer_default.port)
|
||||
process.kill(process.pid, 'SIGTERM')
|
||||
}
|
||||
|
||||
tcpconsumer_default.registerPacketProcessor(async function (packet) {
|
||||
return new Promise((resolve) => {
|
||||
packet.payload = packet.payload +':local'
|
||||
resolve(packet)})
|
||||
})
|
||||
|
||||
let packet = {payload:'tcp payload'}
|
||||
let res = await tcpconsumer_default.send(packet)
|
||||
expect(res.payload).to.equal('8080:tcp payload:local')
|
||||
|
||||
}) // end tcp socket test
|
||||
|
||||
it('with alternate port, and registered consumer processor', async function () {
|
||||
|
||||
let tcpconsumer_9080 = new Consumer({port:9080, name:'tcp-consumer-9080'})
|
||||
|
||||
|
||||
let [err] = await btc(tcpconsumer_9080.connect)()
|
||||
if (err) {
|
||||
console.log('unable to connect to socket to start test', tcpconsumer_9080.port)
|
||||
process.kill(process.pid, 'SIGTERM')
|
||||
}
|
||||
|
||||
tcpconsumer_9080.registerPacketProcessor(async function (packet) {
|
||||
return new Promise((resolve) => {
|
||||
packet.payload = packet.payload +':local'
|
||||
resolve(packet)})
|
||||
})
|
||||
|
||||
let packet = {payload:'tcp payload'}
|
||||
let res = await tcpconsumer_9080.send(packet)
|
||||
expect(res.payload).to.equal('9080:tcp payload:local')
|
||||
|
||||
}) // end tcp socket 2 test
|
||||
|
||||
|
||||
}) // end describe
|
|
@ -1,83 +0,0 @@
|
|||
import { spawn } from 'child_process'
|
||||
import chai from 'chai'
|
||||
import chaiAsPromised from 'chai-as-promised'
|
||||
import btc from 'better-try-catch'
|
||||
chai.use(chaiAsPromised)
|
||||
const expect = chai.expect
|
||||
|
||||
import { Consumer } from '../src'
|
||||
|
||||
const SOCKET_FILE = 'usocket-default'
|
||||
|
||||
let consumer = new Consumer({path:true,name:'unix-consumer'})
|
||||
let consumer2 = new Consumer({path:true, name:'unix-consumer2'})
|
||||
|
||||
let socket = {}
|
||||
|
||||
describe('Connects and Processes a payload via Unix Socket using JSON packet with defaults', function(){
|
||||
|
||||
before(async function(){
|
||||
socket = spawn('node',['-r', '@std/esm', './test/sockets/'+SOCKET_FILE])
|
||||
socket.stdout.on('data', function(buf) {
|
||||
console.log('[Socket]', String(buf))
|
||||
})
|
||||
})
|
||||
|
||||
after(async function(){
|
||||
socket.kill()
|
||||
})
|
||||
|
||||
const TIMES = 3000
|
||||
|
||||
it(`Tests unix socket with default echo JSON packet procssing with ${TIMES} packets sent`, async function () {
|
||||
|
||||
let [err] = await btc(consumer.connect)()
|
||||
if (err) {
|
||||
console.log('unable to connect to socket to start test', consumer.path)
|
||||
process.kill(process.pid, 'SIGTERM')
|
||||
}
|
||||
|
||||
consumer.registerPacketProcessor(async function (packet) {
|
||||
return new Promise((resolve) => {
|
||||
packet.times += 1
|
||||
if (packet.times === TIMES) packet.payload = 'local1:'+packet.payload
|
||||
resolve(packet)})
|
||||
})
|
||||
|
||||
|
||||
let packet = {payload:'payload', times:0}
|
||||
for (let i = 1; i <= TIMES; i++) {
|
||||
packet = await consumer.send(packet)
|
||||
}
|
||||
expect(packet.payload+':'+packet.times).to.equal('local1:payload:'+TIMES)
|
||||
|
||||
}) // end unix socket test
|
||||
|
||||
|
||||
it(`unix socket with two consumers alternating packets, ${TIMES} packets each and local processing`, async function () {
|
||||
|
||||
|
||||
let [err] = await btc(consumer2.connect)()
|
||||
if (err) {
|
||||
console.log('unable to connect to socket to start test', consumer.path)
|
||||
process.kill(process.pid, 'SIGTERM')
|
||||
}
|
||||
|
||||
consumer2.registerPacketProcessor(async function (packet) {
|
||||
return new Promise((resolve) => {
|
||||
packet.times += 1
|
||||
if (packet.times === TIMES) packet.payload = 'local2:'+packet.payload
|
||||
resolve(packet)})
|
||||
})
|
||||
|
||||
let packet = {consumer:1, payload:'payload', times:-1}
|
||||
for (let i = 0; i < TIMES; i++) {
|
||||
packet = await consumer.send(packet)
|
||||
if (packet.times === TIMES) packet.times = 1
|
||||
packet = await consumer2.send(packet)
|
||||
}
|
||||
expect(packet.payload+':'+packet.times).to.equal('local2:local1:payload:'+TIMES)
|
||||
|
||||
}) // end unix socket test
|
||||
|
||||
}) // end describe
|
|
@ -1,51 +0,0 @@
|
|||
import { spawn } from 'child_process'
|
||||
import chai from 'chai'
|
||||
import chaiAsPromised from 'chai-as-promised'
|
||||
import btc from 'better-try-catch'
|
||||
chai.use(chaiAsPromised)
|
||||
const expect = chai.expect
|
||||
|
||||
import { Consumer } from '../src'
|
||||
|
||||
const SOCKET_FILE = 'usocket'
|
||||
|
||||
let consumer = new Consumer({path:SOCKET_FILE,name:'unix-consumer'})
|
||||
|
||||
// const delay = time => new Promise(res=>setTimeout(()=>res(),time))
|
||||
|
||||
let socket = {}
|
||||
|
||||
describe('Connects and Processes a payload via Unix Socket using JSON packet with alt processor', function(){
|
||||
|
||||
before(async function(){
|
||||
socket = spawn('node',['-r', '@std/esm', './test/sockets/'+SOCKET_FILE])
|
||||
socket.stdout.on('data', function(buf) {
|
||||
console.log('[Socket]', String(buf))
|
||||
})
|
||||
})
|
||||
|
||||
after(async function(){
|
||||
socket.kill()
|
||||
})
|
||||
|
||||
it('Tests alternate JSON packet procssing at socket and consumer', async function () {
|
||||
|
||||
let [err] = await btc(consumer.connect)()
|
||||
if (err) {
|
||||
console.log('unable to connect to socket to start test', consumer.path)
|
||||
process.kill(process.pid, 'SIGTERM')
|
||||
}
|
||||
|
||||
consumer.registerPacketProcessor(async function (packet) {
|
||||
return new Promise((resolve) => {
|
||||
packet.payload = 'local:'+packet.payload
|
||||
resolve(packet)})
|
||||
})
|
||||
let packet = { payload:'payload'}
|
||||
let res = await consumer.send(packet)
|
||||
expect(res.payload).to.equal('local:overwrite default processor from instance at socket => payload')
|
||||
|
||||
|
||||
}) // end unix socket test
|
||||
|
||||
}) // end describe
|
Loading…
Reference in New Issue