diff --git a/.eslintrc.js b/.eslintrc.js index 2bed546..d4dda45 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -9,9 +9,11 @@ module.exports = { "node": true, "mocha": true }, + "parser": "babel-eslint", "parserOptions": { "ecmaVersion": 2017, - "sourceType": "module" + "sourceType": "module", + "allowImportExportEverywhere": true }, "extends": "eslint:recommended", "rules": { diff --git a/examples/server.js b/examples/server.js index b037928..55f6c01 100644 --- a/examples/server.js +++ b/examples/server.js @@ -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 () => { + // TODO dynamic import + if(TLS_KEY_PATH && TLS_CRT_PATH && TLS) { + Socket = sSocket + console.log('using TLS') + } class Test extends Socket { 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({path:true}) + let test = new Test(options) await test.create() })().catch(err => { diff --git a/package.json b/package.json index e2352dc..35d81d5 100644 --- a/package.json +++ b/package.json @@ -6,14 +6,14 @@ "scripts": { "testw": "mocha -r esm test/*.test.mjs --watch --recurse --watch-extensions 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", - "s": "DEBUG=true node -r esm examples/server", - "sp": "DEBUG=true node -r esm examples/server-push", - "devs": "SOCKETS_DIR=/opt/sockets DEBUG=true ./node_modules/.bin/nodemon -r esm-e mjs examples/server", - "c": "DEBUG=true node -r esm examples/client", - "cp": "DEBUG=true node -r esm examples/client-push", - "devc": "SOCKETS_DIR=/opt/sockets DEBUG=true node -r esm examples/client", + "s": "UCI_DEV=true node -r esm examples/server", + "sp": "UCI_DEV=true node -r esm examples/server-push", + "devs": "SOCKETS_DIR=/opt/sockets UCI_DEV=true ./node_modules/.bin/nodemon -r esm-e mjs examples/server", + "c": "UCI_DEV=true node -r esm examples/client", + "cp": "UCI_DEV=true node -r esm examples/client-push", + "devc": "SOCKETS_DIR=/opt/sockets UCI_DEV=true node -r esm examples/client", "c2": "node -r esm examples/client2" }, "author": "David Kebler", @@ -44,6 +44,7 @@ "esm": "^3.0.84", "istanbul": "^0.4.5", "mocha": "^5.2.0", + "mz": "^2.7.0", "nodemon": "^1.18.6" }, "dependencies": { diff --git a/src/index.js b/src/index.js index 55c4e20..1d77f13 100644 --- a/src/index.js +++ b/src/index.js @@ -1,6 +1,8 @@ import Socket from './socket' +import sSocket from './sSocket' import Consumer from './consumer' +export { sSocket as sSocket } export { Socket as Socket } export { Consumer as Consumer } -export default { Socket, Consumer } +export default { Socket, sSocket, Consumer } diff --git a/src/socket.js b/src/socket.js index 5e7d5e1..bea258a 100644 --- a/src/socket.js +++ b/src/socket.js @@ -1,258 +1,5 @@ -// node modules +// return an unencrypted/unsecure socket class import { Server } from 'net' -import { unlink as fileDelete } from 'fs' -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 +import socketClass from './socket-class' -// TODO change default pipe dir depending on OS linux,windows,mac -/** @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 +export default (() => { return socketClass(Server) })()