1
  2/**
  3 * Module dependencies.
  4 */
  5
  6var Emitter = require('events').EventEmitter;
  7var parser = require('socket.io-parser');
  8var url = require('url');
  9var debug = require('debug')('socket.io:socket');
 10var hasBin = require('has-binary');
 11var assign = require('object-assign');
 12
 13/**
 14 * Module exports.
 15 */
 16
 17module.exports = exports = Socket;
 18
 19/**
 20 * Blacklisted events.
 21 *
 22 * @api public
 23 */
 24
 25exports.events = [
 26  'error',
 27  'connect',
 28  'disconnect',
 29  'disconnecting',
 30  'newListener',
 31  'removeListener'
 32];
 33
 34/**
 35 * Flags.
 36 *
 37 * @api private
 38 */
 39
 40var flags = [
 41  'json',
 42  'volatile',
 43  'broadcast'
 44];
 45
 46/**
 47 * `EventEmitter#emit` reference.
 48 */
 49
 50var emit = Emitter.prototype.emit;
 51
 52/**
 53 * Interface to a `Client` for a given `Namespace`.
 54 *
 55 * @param {Namespace} nsp
 56 * @param {Client} client
 57 * @api public
 58 */
 59
 60function Socket(nsp, client, query){
 61  this.nsp = nsp;
 62  this.server = nsp.server;
 63  this.adapter = this.nsp.adapter;
 64  this.id = nsp.name !== '/' ? nsp.name + '#' + client.id : client.id;
 65  this.client = client;
 66  this.conn = client.conn;
 67  this.rooms = {};
 68  this.acks = {};
 69  this.connected = true;
 70  this.disconnected = false;
 71  this.handshake = this.buildHandshake(query);
 72  this.fns = [];
 73}
 74
 75/**
 76 * Inherits from `EventEmitter`.
 77 */
 78
 79Socket.prototype.__proto__ = Emitter.prototype;
 80
 81/**
 82 * Apply flags from `Socket`.
 83 */
 84
 85flags.forEach(function(flag){
 86  Object.defineProperty(Socket.prototype, flag, {
 87    get: function() {
 88      this.flags = this.flags || {};
 89      this.flags[flag] = true;
 90      return this;
 91    }
 92  });
 93});
 94
 95/**
 96 * `request` engine.io shortcut.
 97 *
 98 * @api public
 99 */
100
101Object.defineProperty(Socket.prototype, 'request', {
102  get: function() {
103    return this.conn.request;
104  }
105});
106
107/**
108 * Builds the `handshake` BC object
109 *
110 * @api private
111 */
112
113Socket.prototype.buildHandshake = function(query){
114  var self = this;
115  function buildQuery(){
116    var requestQuery = url.parse(self.request.url, true).query;
117    //if socket-specific query exist, replace query strings in requestQuery
118    return assign({}, query, requestQuery);
119  }
120  return {
121    headers: this.request.headers,
122    time: (new Date) + '',
123    address: this.conn.remoteAddress,
124    xdomain: !!this.request.headers.origin,
125    secure: !!this.request.connection.encrypted,
126    issued: +(new Date),
127    url: this.request.url,
128    query: buildQuery()
129  };
130};
131
132/**
133 * Emits to this client.
134 *
135 * @return {Socket} self
136 * @api public
137 */
138
139Socket.prototype.emit = function(ev){
140  if (~exports.events.indexOf(ev)) {
141    emit.apply(this, arguments);
142  } else {
143    var args = Array.prototype.slice.call(arguments);
144    var packet = {};
145    packet.type = hasBin(args) ? parser.BINARY_EVENT : parser.EVENT;
146    packet.data = args;
147    var flags = this.flags || {};
148
149    // access last argument to see if it's an ACK callback
150    if ('function' == typeof args[args.length - 1]) {
151      if (this._rooms || flags.broadcast) {
152        throw new Error('Callbacks are not supported when broadcasting');
153      }
154
155      debug('emitting packet with ack id %d', this.nsp.ids);
156      this.acks[this.nsp.ids] = args.pop();
157      packet.id = this.nsp.ids++;
158    }
159
160    if (this._rooms || flags.broadcast) {
161      this.adapter.broadcast(packet, {
162        except: [this.id],
163        rooms: this._rooms,
164        flags: flags
165      });
166    } else {
167      // dispatch packet
168      this.packet(packet, {
169        volatile: flags.volatile,
170        compress: flags.compress
171      });
172    }
173
174    // reset flags
175    delete this._rooms;
176    delete this.flags;
177  }
178  return this;
179};
180
181/**
182 * Targets a room when broadcasting.
183 *
184 * @param {String} name
185 * @return {Socket} self
186 * @api public
187 */
188
189Socket.prototype.to =
190Socket.prototype.in = function(name){
191  this._rooms = this._rooms || [];
192  if (!~this._rooms.indexOf(name)) this._rooms.push(name);
193  return this;
194};
195
196/**
197 * Sends a `message` event.
198 *
199 * @return {Socket} self
200 * @api public
201 */
202
203Socket.prototype.send =
204Socket.prototype.write = function(){
205  var args = Array.prototype.slice.call(arguments);
206  args.unshift('message');
207  this.emit.apply(this, args);
208  return this;
209};
210
211/**
212 * Writes a packet.
213 *
214 * @param {Object} packet object
215 * @param {Object} opts options
216 * @api private
217 */
218
219Socket.prototype.packet = function(packet, opts){
220  packet.nsp = this.nsp.name;
221  opts = opts || {};
222  opts.compress = false !== opts.compress;
223  this.client.packet(packet, opts);
224};
225
226/**
227 * Joins a room.
228 *
229 * @param {String} room
230 * @param {Function} fn optional, callback
231 * @return {Socket} self
232 * @api private
233 */
234
235Socket.prototype.join = function(room, fn){
236  debug('joining room %s', room);
237  var self = this;
238  if (this.rooms.hasOwnProperty(room)) {
239    fn && fn(null);
240    return this;
241  }
242  this.adapter.add(this.id, room, function(err){
243    if (err) return fn && fn(err);
244    debug('joined room %s', room);
245    self.rooms[room] = room;
246    fn && fn(null);
247  });
248  return this;
249};
250
251/**
252 * Leaves a room.
253 *
254 * @param {String} room
255 * @param {Function} fn optional, callback
256 * @return {Socket} self
257 * @api private
258 */
259
260Socket.prototype.leave = function(room, fn){
261  debug('leave room %s', room);
262  var self = this;
263  this.adapter.del(this.id, room, function(err){
264    if (err) return fn && fn(err);
265    debug('left room %s', room);
266    delete self.rooms[room];
267    fn && fn(null);
268  });
269  return this;
270};
271
272/**
273 * Leave all rooms.
274 *
275 * @api private
276 */
277
278Socket.prototype.leaveAll = function(){
279  this.adapter.delAll(this.id);
280  this.rooms = {};
281};
282
283/**
284 * Called by `Namespace` upon successful
285 * middleware execution (ie: authorization).
286 * Socket is added to namespace array before
287 * call to join, so adapters can access it.
288 *
289 * @api private
290 */
291
292Socket.prototype.onconnect = function(){
293  debug('socket connected - writing packet');
294  this.nsp.connected[this.id] = this;
295  this.join(this.id);
296  this.packet({ type: parser.CONNECT });
297};
298
299/**
300 * Called with each packet. Called by `Client`.
301 *
302 * @param {Object} packet
303 * @api private
304 */
305
306Socket.prototype.onpacket = function(packet){
307  debug('got packet %j', packet);
308  switch (packet.type) {
309    case parser.EVENT:
310      this.onevent(packet);
311      break;
312
313    case parser.BINARY_EVENT:
314      this.onevent(packet);
315      break;
316
317    case parser.ACK:
318      this.onack(packet);
319      break;
320
321    case parser.BINARY_ACK:
322      this.onack(packet);
323      break;
324
325    case parser.DISCONNECT:
326      this.ondisconnect();
327      break;
328
329    case parser.ERROR:
330      this.emit('error', packet.data);
331  }
332};
333
334/**
335 * Called upon event packet.
336 *
337 * @param {Object} packet object
338 * @api private
339 */
340
341Socket.prototype.onevent = function(packet){
342  var args = packet.data || [];
343  debug('emitting event %j', args);
344
345  if (null != packet.id) {
346    debug('attaching ack callback to event');
347    args.push(this.ack(packet.id));
348  }
349
350  this.dispatch(args);
351};
352
353/**
354 * Produces an ack callback to emit with an event.
355 *
356 * @param {Number} id packet id
357 * @api private
358 */
359
360Socket.prototype.ack = function(id){
361  var self = this;
362  var sent = false;
363  return function(){
364    // prevent double callbacks
365    if (sent) return;
366    var args = Array.prototype.slice.call(arguments);
367    debug('sending ack %j', args);
368
369    var type = hasBin(args) ? parser.BINARY_ACK : parser.ACK;
370    self.packet({
371      id: id,
372      type: type,
373      data: args
374    });
375
376    sent = true;
377  };
378};
379
380/**
381 * Called upon ack packet.
382 *
383 * @api private
384 */
385
386Socket.prototype.onack = function(packet){
387  var ack = this.acks[packet.id];
388  if ('function' == typeof ack) {
389    debug('calling ack %s with %j', packet.id, packet.data);
390    ack.apply(this, packet.data);
391    delete this.acks[packet.id];
392  } else {
393    debug('bad ack %s', packet.id);
394  }
395};
396
397/**
398 * Called upon client disconnect packet.
399 *
400 * @api private
401 */
402
403Socket.prototype.ondisconnect = function(){
404  debug('got disconnect packet');
405  this.onclose('client namespace disconnect');
406};
407
408/**
409 * Handles a client error.
410 *
411 * @api private
412 */
413
414Socket.prototype.onerror = function(err){
415  if (this.listeners('error').length) {
416    this.emit('error', err);
417  } else {
418    console.error('Missing error handler on `socket`.');
419    console.error(err.stack);
420  }
421};
422
423/**
424 * Called upon closing. Called by `Client`.
425 *
426 * @param {String} reason
427 * @throw {Error} optional error object
428 * @api private
429 */
430
431Socket.prototype.onclose = function(reason){
432  if (!this.connected) return this;
433  debug('closing socket - reason %s', reason);
434  this.emit('disconnecting', reason);
435  this.leaveAll();
436  this.nsp.remove(this);
437  this.client.remove(this);
438  this.connected = false;
439  this.disconnected = true;
440  delete this.nsp.connected[this.id];
441  this.emit('disconnect', reason);
442};
443
444/**
445 * Produces an `error` packet.
446 *
447 * @param {Object} err error object
448 * @api private
449 */
450
451Socket.prototype.error = function(err){
452  this.packet({ type: parser.ERROR, data: err });
453};
454
455/**
456 * Disconnects this client.
457 *
458 * @param {Boolean} close if `true`, closes the underlying connection
459 * @return {Socket} self
460 * @api public
461 */
462
463Socket.prototype.disconnect = function(close){
464  if (!this.connected) return this;
465  if (close) {
466    this.client.disconnect();
467  } else {
468    this.packet({ type: parser.DISCONNECT });
469    this.onclose('server namespace disconnect');
470  }
471  return this;
472};
473
474/**
475 * Sets the compress flag.
476 *
477 * @param {Boolean} compress if `true`, compresses the sending data
478 * @return {Socket} self
479 * @api public
480 */
481
482Socket.prototype.compress = function(compress){
483  this.flags = this.flags || {};
484  this.flags.compress = compress;
485  return this;
486};
487
488/**
489 * Dispatch incoming event to socket listeners.
490 *
491 * @param {Array} event that will get emitted
492 * @api private
493 */
494
495Socket.prototype.dispatch = function(event){
496  debug('dispatching an event %j', event);
497  var self = this;
498  this.run(event, function(err){
499    process.nextTick(function(){
500      if (err) {
501        return self.error(err.data || err.message);
502      }
503      emit.apply(self, event);
504    });
505  });
506}
507
508/**
509 * Sets up socket middleware.
510 *
511 * @param {Function} middleware function (event, next)
512 * @return {Socket} self
513 * @api public
514 */
515
516Socket.prototype.use = function(fn){
517  this.fns.push(fn);
518  return this;
519};
520
521/**
522 * Executes the middleware for an incoming event.
523 *
524 * @param {Array} event that will get emitted
525 * @param {Function} last fn call in the middleware
526 * @api private
527 */
528Socket.prototype.run = function(event, fn){
529  var fns = this.fns.slice(0);
530  if (!fns.length) return fn(null);
531
532  function run(i){
533    fns[i](event, function(err){
534      // upon error, short-circuit
535      if (err) return fn(err);
536
537      // if no middleware left, summon callback
538      if (!fns[i + 1]) return fn(null);
539
540      // go on to next
541      run(i + 1);
542    });
543  }
544
545  run(0);
546};