twitst4tz

socket.js (11301B)

---

 
 /**
  * Module dependencies.
  */
 
 var Emitter = require('events').EventEmitter;
 var parser = require('socket.io-parser');
 var hasBin = require('has-binary2');
 var url = require('url');
 var debug = require('debug')('socket.io:socket');
 
 /**
  * Module exports.
  */
 
 module.exports = exports = Socket;
 
 /**
  * Blacklisted events.
  *
  * @api public
  */
 
 exports.events = [
   'error',
   'connect',
   'disconnect',
   'disconnecting',
   'newListener',
   'removeListener'
 ];
 
 /**
  * Flags.
  *
  * @api private
  */
 
 var flags = [
   'json',
   'volatile',
   'broadcast',
   'local'
 ];
 
 /**
  * `EventEmitter#emit` reference.
  */
 
 var emit = Emitter.prototype.emit;
 
 /**
  * Interface to a `Client` for a given `Namespace`.
  *
  * @param {Namespace} nsp
  * @param {Client} client
  * @api public
  */
 
 function Socket(nsp, client, query){
   this.nsp = nsp;
   this.server = nsp.server;
   this.adapter = this.nsp.adapter;
   this.id = nsp.name !== '/' ? nsp.name + '#' + client.id : client.id;
   this.client = client;
   this.conn = client.conn;
   this.rooms = {};
   this.acks = {};
   this.connected = true;
   this.disconnected = false;
   this.handshake = this.buildHandshake(query);
   this.fns = [];
   this.flags = {};
   this._rooms = [];
 }
 
 /**
  * Inherits from `EventEmitter`.
  */
 
 Socket.prototype.__proto__ = Emitter.prototype;
 
 /**
  * Apply flags from `Socket`.
  */
 
 flags.forEach(function(flag){
   Object.defineProperty(Socket.prototype, flag, {
     get: function() {
       this.flags[flag] = true;
       return this;
     }
   });
 });
 
 /**
  * `request` engine.io shortcut.
  *
  * @api public
  */
 
 Object.defineProperty(Socket.prototype, 'request', {
   get: function() {
     return this.conn.request;
   }
 });
 
 /**
  * Builds the `handshake` BC object
  *
  * @api private
  */
 
 Socket.prototype.buildHandshake = function(query){
   var self = this;
   function buildQuery(){
     var requestQuery = url.parse(self.request.url, true).query;
     //if socket-specific query exist, replace query strings in requestQuery
     return Object.assign({}, query, requestQuery);
   }
   return {
     headers: this.request.headers,
     time: (new Date) + '',
     address: this.conn.remoteAddress,
     xdomain: !!this.request.headers.origin,
     secure: !!this.request.connection.encrypted,
     issued: +(new Date),
     url: this.request.url,
     query: buildQuery()
   };
 };
 
 /**
  * Emits to this client.
  *
  * @return {Socket} self
  * @api public
  */
 
 Socket.prototype.emit = function(ev){
   if (~exports.events.indexOf(ev)) {
     emit.apply(this, arguments);
     return this;
   }
 
   var args = Array.prototype.slice.call(arguments);
   var packet = {
     type: (this.flags.binary !== undefined ? this.flags.binary : hasBin(args)) ? parser.BINARY_EVENT : parser.EVENT,
     data: args
   };
 
   // access last argument to see if it's an ACK callback
   if (typeof args[args.length - 1] === 'function') {
     if (this._rooms.length || this.flags.broadcast) {
       throw new Error('Callbacks are not supported when broadcasting');
     }
 
     debug('emitting packet with ack id %d', this.nsp.ids);
     this.acks[this.nsp.ids] = args.pop();
     packet.id = this.nsp.ids++;
   }
 
   var rooms = this._rooms.slice(0);
   var flags = Object.assign({}, this.flags);
 
   // reset flags
   this._rooms = [];
   this.flags = {};
 
   if (rooms.length || flags.broadcast) {
     this.adapter.broadcast(packet, {
       except: [this.id],
       rooms: rooms,
       flags: flags
     });
   } else {
     // dispatch packet
     this.packet(packet, flags);
   }
   return this;
 };
 
 /**
  * Targets a room when broadcasting.
  *
  * @param {String} name
  * @return {Socket} self
  * @api public
  */
 
 Socket.prototype.to =
 Socket.prototype.in = function(name){
   if (!~this._rooms.indexOf(name)) this._rooms.push(name);
   return this;
 };
 
 /**
  * Sends a `message` event.
  *
  * @return {Socket} self
  * @api public
  */
 
 Socket.prototype.send =
 Socket.prototype.write = function(){
   var args = Array.prototype.slice.call(arguments);
   args.unshift('message');
   this.emit.apply(this, args);
   return this;
 };
 
 /**
  * Writes a packet.
  *
  * @param {Object} packet object
  * @param {Object} opts options
  * @api private
  */
 
 Socket.prototype.packet = function(packet, opts){
   packet.nsp = this.nsp.name;
   opts = opts || {};
   opts.compress = false !== opts.compress;
   this.client.packet(packet, opts);
 };
 
 /**
  * Joins a room.
  *
  * @param {String|Array} room or array of rooms
  * @param {Function} fn optional, callback
  * @return {Socket} self
  * @api private
  */
 
 Socket.prototype.join = function(rooms, fn){
   debug('joining room %s', rooms);
   var self = this;
   if (!Array.isArray(rooms)) {
     rooms = [rooms];
   }
   rooms = rooms.filter(function (room) {
     return !self.rooms.hasOwnProperty(room);
   });
   if (!rooms.length) {
     fn && fn(null);
     return this;
   }
   this.adapter.addAll(this.id, rooms, function(err){
     if (err) return fn && fn(err);
     debug('joined room %s', rooms);
     rooms.forEach(function (room) {
       self.rooms[room] = room;
     });
     fn && fn(null);
   });
   return this;
 };
 
 /**
  * Leaves a room.
  *
  * @param {String} room
  * @param {Function} fn optional, callback
  * @return {Socket} self
  * @api private
  */
 
 Socket.prototype.leave = function(room, fn){
   debug('leave room %s', room);
   var self = this;
   this.adapter.del(this.id, room, function(err){
     if (err) return fn && fn(err);
     debug('left room %s', room);
     delete self.rooms[room];
     fn && fn(null);
   });
   return this;
 };
 
 /**
  * Leave all rooms.
  *
  * @api private
  */
 
 Socket.prototype.leaveAll = function(){
   this.adapter.delAll(this.id);
   this.rooms = {};
 };
 
 /**
  * Called by `Namespace` upon successful
  * middleware execution (ie: authorization).
  * Socket is added to namespace array before
  * call to join, so adapters can access it.
  *
  * @api private
  */
 
 Socket.prototype.onconnect = function(){
   debug('socket connected - writing packet');
   this.nsp.connected[this.id] = this;
   this.join(this.id);
   var skip = this.nsp.name === '/' && this.nsp.fns.length === 0;
   if (skip) {
     debug('packet already sent in initial handshake');
   } else {
     this.packet({ type: parser.CONNECT });
   }
 };
 
 /**
  * Called with each packet. Called by `Client`.
  *
  * @param {Object} packet
  * @api private
  */
 
 Socket.prototype.onpacket = function(packet){
   debug('got packet %j', packet);
   switch (packet.type) {
     case parser.EVENT:
       this.onevent(packet);
       break;
 
     case parser.BINARY_EVENT:
       this.onevent(packet);
       break;
 
     case parser.ACK:
       this.onack(packet);
       break;
 
     case parser.BINARY_ACK:
       this.onack(packet);
       break;
 
     case parser.DISCONNECT:
       this.ondisconnect();
       break;
 
     case parser.ERROR:
       this.onerror(new Error(packet.data));
   }
 };
 
 /**
  * Called upon event packet.
  *
  * @param {Object} packet object
  * @api private
  */
 
 Socket.prototype.onevent = function(packet){
   var args = packet.data || [];
   debug('emitting event %j', args);
 
   if (null != packet.id) {
     debug('attaching ack callback to event');
     args.push(this.ack(packet.id));
   }
 
   this.dispatch(args);
 };
 
 /**
  * Produces an ack callback to emit with an event.
  *
  * @param {Number} id packet id
  * @api private
  */
 
 Socket.prototype.ack = function(id){
   var self = this;
   var sent = false;
   return function(){
     // prevent double callbacks
     if (sent) return;
     var args = Array.prototype.slice.call(arguments);
     debug('sending ack %j', args);
 
     self.packet({
       id: id,
       type: hasBin(args) ? parser.BINARY_ACK : parser.ACK,
       data: args
     });
 
     sent = true;
   };
 };
 
 /**
  * Called upon ack packet.
  *
  * @api private
  */
 
 Socket.prototype.onack = function(packet){
   var ack = this.acks[packet.id];
   if ('function' == typeof ack) {
     debug('calling ack %s with %j', packet.id, packet.data);
     ack.apply(this, packet.data);
     delete this.acks[packet.id];
   } else {
     debug('bad ack %s', packet.id);
   }
 };
 
 /**
  * Called upon client disconnect packet.
  *
  * @api private
  */
 
 Socket.prototype.ondisconnect = function(){
   debug('got disconnect packet');
   this.onclose('client namespace disconnect');
 };
 
 /**
  * Handles a client error.
  *
  * @api private
  */
 
 Socket.prototype.onerror = function(err){
   if (this.listeners('error').length) {
     this.emit('error', err);
   } else {
     console.error('Missing error handler on `socket`.');
     console.error(err.stack);
   }
 };
 
 /**
  * Called upon closing. Called by `Client`.
  *
  * @param {String} reason
  * @throw {Error} optional error object
  * @api private
  */
 
 Socket.prototype.onclose = function(reason){
   if (!this.connected) return this;
   debug('closing socket - reason %s', reason);
   this.emit('disconnecting', reason);
   this.leaveAll();
   this.nsp.remove(this);
   this.client.remove(this);
   this.connected = false;
   this.disconnected = true;
   delete this.nsp.connected[this.id];
   this.emit('disconnect', reason);
 };
 
 /**
  * Produces an `error` packet.
  *
  * @param {Object} err error object
  * @api private
  */
 
 Socket.prototype.error = function(err){
   this.packet({ type: parser.ERROR, data: err });
 };
 
 /**
  * Disconnects this client.
  *
  * @param {Boolean} close if `true`, closes the underlying connection
  * @return {Socket} self
  * @api public
  */
 
 Socket.prototype.disconnect = function(close){
   if (!this.connected) return this;
   if (close) {
     this.client.disconnect();
   } else {
     this.packet({ type: parser.DISCONNECT });
     this.onclose('server namespace disconnect');
   }
   return this;
 };
 
 /**
  * Sets the compress flag.
  *
  * @param {Boolean} compress if `true`, compresses the sending data
  * @return {Socket} self
  * @api public
  */
 
 Socket.prototype.compress = function(compress){
   this.flags.compress = compress;
   return this;
 };
 
 /**
  * Sets the binary flag
  *
  * @param {Boolean} Encode as if it has binary data if `true`, Encode as if it doesnt have binary data if `false`
  * @return {Socket} self
  * @api public
  */
 
  Socket.prototype.binary = function (binary) {
    this.flags.binary = binary;
    return this;
  };
 
 /**
  * Dispatch incoming event to socket listeners.
  *
  * @param {Array} event that will get emitted
  * @api private
  */
 
 Socket.prototype.dispatch = function(event){
   debug('dispatching an event %j', event);
   var self = this;
   function dispatchSocket(err) {
     process.nextTick(function(){
       if (err) {
         return self.error(err.data || err.message);
       }
       emit.apply(self, event);
     });
   }
   this.run(event, dispatchSocket);
 };
 
 /**
  * Sets up socket middleware.
  *
  * @param {Function} middleware function (event, next)
  * @return {Socket} self
  * @api public
  */
 
 Socket.prototype.use = function(fn){
   this.fns.push(fn);
   return this;
 };
 
 /**
  * Executes the middleware for an incoming event.
  *
  * @param {Array} event that will get emitted
  * @param {Function} last fn call in the middleware
  * @api private
  */
 Socket.prototype.run = function(event, fn){
   var fns = this.fns.slice(0);
   if (!fns.length) return fn(null);
 
   function run(i){
     fns[i](event, function(err){
       // upon error, short-circuit
       if (err) return fn(err);
 
       // if no middleware left, summon callback
       if (!fns[i + 1]) return fn(null);
 
       // go on to next
       run(i + 1);
     });
   }
 
   run(0);
 };