set up socket/server for TLS option
using command Socket Class extending either plain or tls server classtls
parent
63beca4199
commit
9ce3226f93
|
@ -9,9 +9,11 @@ module.exports = {
|
||||||
"node": true,
|
"node": true,
|
||||||
"mocha": true
|
"mocha": true
|
||||||
},
|
},
|
||||||
|
"parser": "babel-eslint",
|
||||||
"parserOptions": {
|
"parserOptions": {
|
||||||
"ecmaVersion": 2017,
|
"ecmaVersion": 2017,
|
||||||
"sourceType": "module"
|
"sourceType": "module",
|
||||||
|
"allowImportExportEverywhere": true
|
||||||
},
|
},
|
||||||
"extends": "eslint:recommended",
|
"extends": "eslint:recommended",
|
||||||
"rules": {
|
"rules": {
|
||||||
|
|
|
@ -1,7 +1,22 @@
|
||||||
import { Socket } from '../src'
|
import { Socket as uSocket, sSocket} from '../src'
|
||||||
|
import { fs } from 'mz'
|
||||||
|
|
||||||
;
|
// made key cert into module that also uses environment variables
|
||||||
|
const TLS = process.env.TLS || false
|
||||||
|
const TLS_DIR = process.env.TLS_DIR || '/opt/certs'
|
||||||
|
const TLS_NAME = process.env.TLD_NAME || 'wc.kebler.net'
|
||||||
|
const TLS_KEY_PATH = process.env.TLS_KEY_PATH || `${TLS_DIR}/${TLS_NAME}.key`
|
||||||
|
const TLS_CRT_PATH = process.env.TLS_CRT_PATH || `${TLS_DIR}/${TLS_NAME}.crt`
|
||||||
|
|
||||||
|
let Socket = uSocket
|
||||||
|
|
||||||
|
;
|
||||||
(async () => {
|
(async () => {
|
||||||
|
// TODO dynamic import
|
||||||
|
if(TLS_KEY_PATH && TLS_CRT_PATH && TLS) {
|
||||||
|
Socket = sSocket
|
||||||
|
console.log('using TLS')
|
||||||
|
}
|
||||||
|
|
||||||
class Test extends Socket {
|
class Test extends Socket {
|
||||||
constructor(opts) {
|
constructor(opts) {
|
||||||
|
@ -29,8 +44,21 @@ import { Socket } from '../src'
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const options = {
|
||||||
|
tls: TLS,
|
||||||
|
key: await fs.readFile(TLS_KEY_PATH),
|
||||||
|
cert: await fs.readFile(TLS_CRT_PATH),
|
||||||
|
// This is necessary only if using client certificate authentication.
|
||||||
|
// requestCert: true,
|
||||||
|
// This is necessary only if the client uses a self-signed certificate.
|
||||||
|
// ca: [ fs.readFileSync('client-cert.pem') ]
|
||||||
|
}
|
||||||
|
|
||||||
|
options.path = true
|
||||||
|
|
||||||
|
|
||||||
// let test = new Test()
|
// let test = new Test()
|
||||||
let test = new Test({path:true})
|
let test = new Test(options)
|
||||||
await test.create()
|
await test.create()
|
||||||
|
|
||||||
})().catch(err => {
|
})().catch(err => {
|
||||||
|
|
15
package.json
15
package.json
|
@ -6,14 +6,14 @@
|
||||||
"scripts": {
|
"scripts": {
|
||||||
"testw": "mocha -r esm test/*.test.mjs --watch --recurse --watch-extensions mjs",
|
"testw": "mocha -r esm test/*.test.mjs --watch --recurse --watch-extensions mjs",
|
||||||
"test": "mocha -r esm --timeout 10000 test/*.test.mjs",
|
"test": "mocha -r esm --timeout 10000 test/*.test.mjs",
|
||||||
"testlog": "DEBUG=true mocha -r esm --timeout 10000 test/*.test.mjs",
|
"testlog": "UCI_DEV=true mocha -r esm --timeout 10000 test/*.test.mjs",
|
||||||
"testci": "istanbul cover ./node_modules/.bin/_mocha --report lcovonly -- -R spec --recursive && codecov || true",
|
"testci": "istanbul cover ./node_modules/.bin/_mocha --report lcovonly -- -R spec --recursive && codecov || true",
|
||||||
"s": "DEBUG=true node -r esm examples/server",
|
"s": "UCI_DEV=true node -r esm examples/server",
|
||||||
"sp": "DEBUG=true node -r esm examples/server-push",
|
"sp": "UCI_DEV=true node -r esm examples/server-push",
|
||||||
"devs": "SOCKETS_DIR=/opt/sockets DEBUG=true ./node_modules/.bin/nodemon -r esm-e mjs examples/server",
|
"devs": "SOCKETS_DIR=/opt/sockets UCI_DEV=true ./node_modules/.bin/nodemon -r esm-e mjs examples/server",
|
||||||
"c": "DEBUG=true node -r esm examples/client",
|
"c": "UCI_DEV=true node -r esm examples/client",
|
||||||
"cp": "DEBUG=true node -r esm examples/client-push",
|
"cp": "UCI_DEV=true node -r esm examples/client-push",
|
||||||
"devc": "SOCKETS_DIR=/opt/sockets DEBUG=true node -r esm examples/client",
|
"devc": "SOCKETS_DIR=/opt/sockets UCI_DEV=true node -r esm examples/client",
|
||||||
"c2": "node -r esm examples/client2"
|
"c2": "node -r esm examples/client2"
|
||||||
},
|
},
|
||||||
"author": "David Kebler",
|
"author": "David Kebler",
|
||||||
|
@ -44,6 +44,7 @@
|
||||||
"esm": "^3.0.84",
|
"esm": "^3.0.84",
|
||||||
"istanbul": "^0.4.5",
|
"istanbul": "^0.4.5",
|
||||||
"mocha": "^5.2.0",
|
"mocha": "^5.2.0",
|
||||||
|
"mz": "^2.7.0",
|
||||||
"nodemon": "^1.18.6"
|
"nodemon": "^1.18.6"
|
||||||
},
|
},
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
|
|
|
@ -1,6 +1,8 @@
|
||||||
import Socket from './socket'
|
import Socket from './socket'
|
||||||
|
import sSocket from './sSocket'
|
||||||
import Consumer from './consumer'
|
import Consumer from './consumer'
|
||||||
|
|
||||||
|
export { sSocket as sSocket }
|
||||||
export { Socket as Socket }
|
export { Socket as Socket }
|
||||||
export { Consumer as Consumer }
|
export { Consumer as Consumer }
|
||||||
export default { Socket, Consumer }
|
export default { Socket, sSocket, Consumer }
|
||||||
|
|
259
src/socket.js
259
src/socket.js
|
@ -1,258 +1,5 @@
|
||||||
// node modules
|
// return an unencrypted/unsecure socket class
|
||||||
import { Server } from 'net'
|
import { Server } from 'net'
|
||||||
import { unlink as fileDelete } from 'fs'
|
import socketClass from './socket-class'
|
||||||
import { promisify } from 'util'
|
|
||||||
import path from 'path'
|
|
||||||
// npmjs modules
|
|
||||||
import mkdir from 'make-dir'
|
|
||||||
import btc from 'better-try-catch'
|
|
||||||
import _ON_DEATH from 'death' //this is intentionally ugly
|
|
||||||
import JSONStream from './json-stream'
|
|
||||||
import clone from 'clone'
|
|
||||||
// uci modules
|
|
||||||
import logger from '@uci/logger'
|
|
||||||
let log = {} // must declare here and set later for module wide access
|
|
||||||
|
|
||||||
// TODO change default pipe dir depending on OS linux,windows,mac
|
export default (() => { return socketClass(Server) })()
|
||||||
/** @constant {String} DEFAULT_PIPE_DIR
|
|
||||||
* @description SOCKETS_DIR environment variable or '/tmp/UCI'
|
|
||||||
*/
|
|
||||||
const DEFAULT_PIPE_DIR = process.env.SOCKETS_DIR || '/tmp/UCI'
|
|
||||||
/** @constant {String} DEFAULT_SOCKET_NAME
|
|
||||||
* @description for named pipe 'uci-sock' if not set in options */
|
|
||||||
const DEFAULT_SOCKET_NAME = 'uci-sock'
|
|
||||||
|
|
||||||
/**
|
|
||||||
* UCI Socket - class used to create a socket (server) that supports passing json packets
|
|
||||||
* supports both named pipes and tcp sockets
|
|
||||||
* also supports push of packets to all connected consumers (clients)
|
|
||||||
* is extended from {@link https://nodejs.org/api/net.html#net_class_net_server | nodejs net.Server }
|
|
||||||
* @extends Server
|
|
||||||
*/
|
|
||||||
|
|
||||||
class Socket extends Server {
|
|
||||||
/**
|
|
||||||
* UCI Socket class constructor
|
|
||||||
* @param {Object} opts hash of options
|
|
||||||
* @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 {Object} options.conPacket A json operson's property
|
|
||||||
*
|
|
||||||
*/
|
|
||||||
constructor(opts = {}) {
|
|
||||||
super()
|
|
||||||
this.id = opts.id || opts.name || 'socket:' + new Date().getTime()
|
|
||||||
if (!opts.path) {
|
|
||||||
opts.host = opts.host || '0.0.0.0'
|
|
||||||
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.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)
|
|
||||||
this.create = this.create.bind(this)
|
|
||||||
log = logger({
|
|
||||||
file: 'src/socket.js',
|
|
||||||
class: 'Socket',
|
|
||||||
name: 'socket',
|
|
||||||
id: this.id
|
|
||||||
})
|
|
||||||
} // end constructor
|
|
||||||
|
|
||||||
/**
|
|
||||||
* create - Description
|
|
||||||
*
|
|
||||||
* @returns {type} Description
|
|
||||||
*/
|
|
||||||
async create() {
|
|
||||||
return new Promise(async (resolve, reject) => {
|
|
||||||
// set up a couple ways to gracefully destroy socket process is killed/aborted
|
|
||||||
_ON_DEATH(async () => {
|
|
||||||
log.info('\nhe\'s dead jim')
|
|
||||||
await this._destroy()
|
|
||||||
})
|
|
||||||
process.once('SIGUSR2', async () => {
|
|
||||||
await this._destroy
|
|
||||||
process.kill(process.pid, 'SIGUSR2')
|
|
||||||
})
|
|
||||||
|
|
||||||
this.once('error', async err => {
|
|
||||||
// recover from socket file that was not removed
|
|
||||||
if (err.code === 'EADDRINUSE') {
|
|
||||||
if (this.opts.path) {
|
|
||||||
// if TCP socket should already be dead
|
|
||||||
let [err, res] = await btc(promisify(fileDelete))(this.opts.path)
|
|
||||||
if (!err) {
|
|
||||||
log.info(
|
|
||||||
{ res: res, socket: this.opts.path },
|
|
||||||
'socket already exists.....deleted'
|
|
||||||
)
|
|
||||||
return await this._listen(this.opts)
|
|
||||||
}
|
|
||||||
log.fatal(
|
|
||||||
{ err: err },
|
|
||||||
'error deleting socket. Can not establish a socket'
|
|
||||||
)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (err.code === 'EACCES') {
|
|
||||||
console.log(
|
|
||||||
{ socket: this.opts.path },
|
|
||||||
'directory does not exist...creating'
|
|
||||||
)
|
|
||||||
await mkdir(path.dirname(this.opts.path))
|
|
||||||
console.log({ socket: this.opts.path }, 'created')
|
|
||||||
log.warn(
|
|
||||||
{ socket: this.opts.path },
|
|
||||||
'directory does not exist...creating'
|
|
||||||
)
|
|
||||||
return await this._listen(this.opts)
|
|
||||||
}
|
|
||||||
// otherwise fatally exit
|
|
||||||
log.info(err, 'creating socket')
|
|
||||||
reject(err)
|
|
||||||
})
|
|
||||||
|
|
||||||
let [err, res] = await btc(this._listen)(this.opts)
|
|
||||||
if (err) reject(err)
|
|
||||||
resolve(res)
|
|
||||||
}) // end creeate promise
|
|
||||||
} // end create
|
|
||||||
|
|
||||||
/**
|
|
||||||
* registerPacketProcessor - Description
|
|
||||||
* @public
|
|
||||||
* @param {func} Description
|
|
||||||
*
|
|
||||||
*/
|
|
||||||
registerPacketProcessor(func) {
|
|
||||||
this._packetProcess = func
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* push - pushes a supplied UCI object packet to all connected clients
|
|
||||||
*
|
|
||||||
* @param {object} packet Description
|
|
||||||
* @param {string} id the header id string of the pushed packet, default: 'pushed'
|
|
||||||
*
|
|
||||||
*/
|
|
||||||
async push(packet, id) {
|
|
||||||
packet._header = { id: id || 'pushed' }
|
|
||||||
log.info(
|
|
||||||
{ opts: this.opts, packet: packet },
|
|
||||||
'pushing a packet to all connected consumers'
|
|
||||||
)
|
|
||||||
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)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
async _listen(opts) {
|
|
||||||
super.listen(opts, async (err, res) => {
|
|
||||||
if (err) return err
|
|
||||||
// this gets called for each client connection and is unique to each
|
|
||||||
this.on('connection', async socket => {
|
|
||||||
const stream = new JSONStream()
|
|
||||||
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 })))
|
|
||||||
if (this.opts.conPacket) {
|
|
||||||
this.opts.conPacket._header = { id: 'pushed' }
|
|
||||||
log.info(
|
|
||||||
{ conPacket: this.opts.conPacket },
|
|
||||||
'pushing a preset command to just connected consumer'
|
|
||||||
)
|
|
||||||
send(await stream.serialize(this.opts.conPacket)) // send a packet command on to consumer on connection
|
|
||||||
}
|
|
||||||
socket.on('data', stream.onData)
|
|
||||||
// TODO need to start error listener for stream so errors can be processed
|
|
||||||
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.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 }
|
|
||||||
})
|
|
||||||
log.info(await send(ser))
|
|
||||||
} // end process message
|
|
||||||
}) // end connecttion consumer
|
|
||||||
log.info({ opts: this.opts }, 'socket created')
|
|
||||||
return res
|
|
||||||
}) // end super listen callback
|
|
||||||
} // end listen
|
|
||||||
|
|
||||||
async _destroy() {
|
|
||||||
log.info('closing down socket')
|
|
||||||
await this.close()
|
|
||||||
log.info('all connections closed....exiting')
|
|
||||||
process.exit()
|
|
||||||
}
|
|
||||||
|
|
||||||
// default packet process, just a simple echo
|
|
||||||
async _packetProcess(packet) {
|
|
||||||
return new Promise(resolve => {
|
|
||||||
resolve(packet)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// must have a consumer socket bound to use
|
|
||||||
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')
|
|
||||||
if (!this.write(packet)) {
|
|
||||||
this.once('drain', cb)
|
|
||||||
} else {
|
|
||||||
process.nextTick(cb)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
} // end class
|
|
||||||
|
|
||||||
export default Socket
|
|
||||||
|
|
Loading…
Reference in New Issue