From 67fdec20726e48ba3a934cb25bb30d47ec4a4f29 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yaroslav=20De=20La=20Pe=C3=B1a=20Smirnov?= Date: Wed, 29 Nov 2017 11:44:34 +0300 Subject: Initial commit, version 0.5.3 --- node_modules/engine.io/lib/socket.js | 486 +++++++++++++++++++++++++++++++++++ 1 file changed, 486 insertions(+) create mode 100644 node_modules/engine.io/lib/socket.js (limited to 'node_modules/engine.io/lib/socket.js') diff --git a/node_modules/engine.io/lib/socket.js b/node_modules/engine.io/lib/socket.js new file mode 100644 index 0000000..177b25c --- /dev/null +++ b/node_modules/engine.io/lib/socket.js @@ -0,0 +1,486 @@ +/** + * Module dependencies. + */ + +var EventEmitter = require('events').EventEmitter; +var util = require('util'); +var debug = require('debug')('engine:socket'); + +/** + * Module exports. + */ + +module.exports = Socket; + +/** + * Client class (abstract). + * + * @api private + */ + +function Socket (id, server, transport, req) { + this.id = id; + this.server = server; + this.upgrading = false; + this.upgraded = false; + this.readyState = 'opening'; + this.writeBuffer = []; + this.packetsFn = []; + this.sentCallbackFn = []; + this.cleanupFn = []; + this.request = req; + + // Cache IP since it might not be in the req later + if (req.websocket && req.websocket._socket) { + this.remoteAddress = req.websocket._socket.remoteAddress; + } else { + this.remoteAddress = req.connection.remoteAddress; + } + + this.checkIntervalTimer = null; + this.upgradeTimeoutTimer = null; + this.pingTimeoutTimer = null; + + this.setTransport(transport); + this.onOpen(); +} + +/** + * Inherits from EventEmitter. + */ + +util.inherits(Socket, EventEmitter); + +/** + * Called upon transport considered open. + * + * @api private + */ + +Socket.prototype.onOpen = function () { + this.readyState = 'open'; + + // sends an `open` packet + this.transport.sid = this.id; + this.sendPacket('open', JSON.stringify({ + sid: this.id, + upgrades: this.getAvailableUpgrades(), + pingInterval: this.server.pingInterval, + pingTimeout: this.server.pingTimeout + })); + + if (this.server.initialPacket) { + this.sendPacket('message', this.server.initialPacket); + } + + this.emit('open'); + this.setPingTimeout(); +}; + +/** + * Called upon transport packet. + * + * @param {Object} packet + * @api private + */ + +Socket.prototype.onPacket = function (packet) { + if ('open' === this.readyState) { + // export packet event + debug('packet'); + this.emit('packet', packet); + + // Reset ping timeout on any packet, incoming data is a good sign of + // other side's liveness + this.setPingTimeout(); + + switch (packet.type) { + case 'ping': + debug('got ping'); + this.sendPacket('pong'); + this.emit('heartbeat'); + break; + + case 'error': + this.onClose('parse error'); + break; + + case 'message': + this.emit('data', packet.data); + this.emit('message', packet.data); + break; + } + } else { + debug('packet received with closed socket'); + } +}; + +/** + * Called upon transport error. + * + * @param {Error} error object + * @api private + */ + +Socket.prototype.onError = function (err) { + debug('transport error'); + this.onClose('transport error', err); +}; + +/** + * Sets and resets ping timeout timer based on client pings. + * + * @api private + */ + +Socket.prototype.setPingTimeout = function () { + var self = this; + clearTimeout(self.pingTimeoutTimer); + self.pingTimeoutTimer = setTimeout(function () { + self.onClose('ping timeout'); + }, self.server.pingInterval + self.server.pingTimeout); +}; + +/** + * Attaches handlers for the given transport. + * + * @param {Transport} transport + * @api private + */ + +Socket.prototype.setTransport = function (transport) { + var onError = this.onError.bind(this); + var onPacket = this.onPacket.bind(this); + var flush = this.flush.bind(this); + var onClose = this.onClose.bind(this, 'transport close'); + + this.transport = transport; + this.transport.once('error', onError); + this.transport.on('packet', onPacket); + this.transport.on('drain', flush); + this.transport.once('close', onClose); + // this function will manage packet events (also message callbacks) + this.setupSendCallback(); + + this.cleanupFn.push(function () { + transport.removeListener('error', onError); + transport.removeListener('packet', onPacket); + transport.removeListener('drain', flush); + transport.removeListener('close', onClose); + }); +}; + +/** + * Upgrades socket to the given transport + * + * @param {Transport} transport + * @api private + */ + +Socket.prototype.maybeUpgrade = function (transport) { + debug('might upgrade socket transport from "%s" to "%s"' + , this.transport.name, transport.name); + + this.upgrading = true; + + var self = this; + + // set transport upgrade timer + self.upgradeTimeoutTimer = setTimeout(function () { + debug('client did not complete upgrade - closing transport'); + cleanup(); + if ('open' === transport.readyState) { + transport.close(); + } + }, this.server.upgradeTimeout); + + function onPacket (packet) { + if ('ping' === packet.type && 'probe' === packet.data) { + transport.send([{ type: 'pong', data: 'probe' }]); + self.emit('upgrading', transport); + clearInterval(self.checkIntervalTimer); + self.checkIntervalTimer = setInterval(check, 100); + } else if ('upgrade' === packet.type && self.readyState !== 'closed') { + debug('got upgrade packet - upgrading'); + cleanup(); + self.transport.discard(); + self.upgraded = true; + self.clearTransport(); + self.setTransport(transport); + self.emit('upgrade', transport); + self.setPingTimeout(); + self.flush(); + if (self.readyState === 'closing') { + transport.close(function () { + self.onClose('forced close'); + }); + } + } else { + cleanup(); + transport.close(); + } + } + + // we force a polling cycle to ensure a fast upgrade + function check () { + if ('polling' === self.transport.name && self.transport.writable) { + debug('writing a noop packet to polling for fast upgrade'); + self.transport.send([{ type: 'noop' }]); + } + } + + function cleanup () { + self.upgrading = false; + + clearInterval(self.checkIntervalTimer); + self.checkIntervalTimer = null; + + clearTimeout(self.upgradeTimeoutTimer); + self.upgradeTimeoutTimer = null; + + transport.removeListener('packet', onPacket); + transport.removeListener('close', onTransportClose); + transport.removeListener('error', onError); + self.removeListener('close', onClose); + } + + function onError (err) { + debug('client did not complete upgrade - %s', err); + cleanup(); + transport.close(); + transport = null; + } + + function onTransportClose () { + onError('transport closed'); + } + + function onClose () { + onError('socket closed'); + } + + transport.on('packet', onPacket); + transport.once('close', onTransportClose); + transport.once('error', onError); + + self.once('close', onClose); +}; + +/** + * Clears listeners and timers associated with current transport. + * + * @api private + */ + +Socket.prototype.clearTransport = function () { + var cleanup; + + var toCleanUp = this.cleanupFn.length; + + for (var i = 0; i < toCleanUp; i++) { + cleanup = this.cleanupFn.shift(); + cleanup(); + } + + // silence further transport errors and prevent uncaught exceptions + this.transport.on('error', function () { + debug('error triggered by discarded transport'); + }); + + // ensure transport won't stay open + this.transport.close(); + + clearTimeout(this.pingTimeoutTimer); +}; + +/** + * Called upon transport considered closed. + * Possible reasons: `ping timeout`, `client error`, `parse error`, + * `transport error`, `server close`, `transport close` + */ + +Socket.prototype.onClose = function (reason, description) { + if ('closed' !== this.readyState) { + this.readyState = 'closed'; + clearTimeout(this.pingTimeoutTimer); + clearInterval(this.checkIntervalTimer); + this.checkIntervalTimer = null; + clearTimeout(this.upgradeTimeoutTimer); + var self = this; + // clean writeBuffer in next tick, so developers can still + // grab the writeBuffer on 'close' event + process.nextTick(function () { + self.writeBuffer = []; + }); + this.packetsFn = []; + this.sentCallbackFn = []; + this.clearTransport(); + this.emit('close', reason, description); + } +}; + +/** + * Setup and manage send callback + * + * @api private + */ + +Socket.prototype.setupSendCallback = function () { + var self = this; + this.transport.on('drain', onDrain); + + this.cleanupFn.push(function () { + self.transport.removeListener('drain', onDrain); + }); + + // the message was sent successfully, execute the callback + function onDrain () { + if (self.sentCallbackFn.length > 0) { + var seqFn = self.sentCallbackFn.splice(0, 1)[0]; + if ('function' === typeof seqFn) { + debug('executing send callback'); + seqFn(self.transport); + } else if (Array.isArray(seqFn)) { + debug('executing batch send callback'); + for (var l = seqFn.length, i = 0; i < l; i++) { + if ('function' === typeof seqFn[i]) { + seqFn[i](self.transport); + } + } + } + } + } +}; + +/** + * Sends a message packet. + * + * @param {String} message + * @param {Object} options + * @param {Function} callback + * @return {Socket} for chaining + * @api public + */ + +Socket.prototype.send = +Socket.prototype.write = function (data, options, callback) { + this.sendPacket('message', data, options, callback); + return this; +}; + +/** + * Sends a packet. + * + * @param {String} packet type + * @param {String} optional, data + * @param {Object} options + * @api private + */ + +Socket.prototype.sendPacket = function (type, data, options, callback) { + if ('function' === typeof options) { + callback = options; + options = null; + } + + options = options || {}; + options.compress = false !== options.compress; + + if ('closing' !== this.readyState && 'closed' !== this.readyState) { + debug('sending packet "%s" (%s)', type, data); + + var packet = { + type: type, + options: options + }; + if (data) packet.data = data; + + // exports packetCreate event + this.emit('packetCreate', packet); + + this.writeBuffer.push(packet); + + // add send callback to object, if defined + if (callback) this.packetsFn.push(callback); + + this.flush(); + } +}; + +/** + * Attempts to flush the packets buffer. + * + * @api private + */ + +Socket.prototype.flush = function () { + if ('closed' !== this.readyState && + this.transport.writable && + this.writeBuffer.length) { + debug('flushing buffer to transport'); + this.emit('flush', this.writeBuffer); + this.server.emit('flush', this, this.writeBuffer); + var wbuf = this.writeBuffer; + this.writeBuffer = []; + if (!this.transport.supportsFraming) { + this.sentCallbackFn.push(this.packetsFn); + } else { + this.sentCallbackFn.push.apply(this.sentCallbackFn, this.packetsFn); + } + this.packetsFn = []; + this.transport.send(wbuf); + this.emit('drain'); + this.server.emit('drain', this); + } +}; + +/** + * Get available upgrades for this socket. + * + * @api private + */ + +Socket.prototype.getAvailableUpgrades = function () { + var availableUpgrades = []; + var allUpgrades = this.server.upgrades(this.transport.name); + for (var i = 0, l = allUpgrades.length; i < l; ++i) { + var upg = allUpgrades[i]; + if (this.server.transports.indexOf(upg) !== -1) { + availableUpgrades.push(upg); + } + } + return availableUpgrades; +}; + +/** + * Closes the socket and underlying transport. + * + * @param {Boolean} optional, discard + * @return {Socket} for chaining + * @api public + */ + +Socket.prototype.close = function (discard) { + if ('open' !== this.readyState) return; + + this.readyState = 'closing'; + + if (this.writeBuffer.length) { + this.once('drain', this.closeTransport.bind(this, discard)); + return; + } + + this.closeTransport(discard); +}; + +/** + * Closes the underlying transport. + * + * @param {Boolean} discard + * @api private + */ + +Socket.prototype.closeTransport = function (discard) { + if (discard) this.transport.discard(); + this.transport.close(this.onClose.bind(this, 'forced close')); +}; -- cgit v1.2.3