twitst4tz

socket.js (11597B)

---

 /**
  * 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'));
 };