1/**
  2 * Module dependencies.
  3 */
  4
  5var EventEmitter = require('events').EventEmitter;
  6var util = require('util');
  7var debug = require('debug')('engine:socket');
  8
  9/**
 10 * Module exports.
 11 */
 12
 13module.exports = Socket;
 14
 15/**
 16 * Client class (abstract).
 17 *
 18 * @api private
 19 */
 20
 21function Socket (id, server, transport, req) {
 22  this.id = id;
 23  this.server = server;
 24  this.upgrading = false;
 25  this.upgraded = false;
 26  this.readyState = 'opening';
 27  this.writeBuffer = [];
 28  this.packetsFn = [];
 29  this.sentCallbackFn = [];
 30  this.cleanupFn = [];
 31  this.request = req;
 32
 33  // Cache IP since it might not be in the req later
 34  this.remoteAddress = req.connection.remoteAddress;
 35
 36  this.checkIntervalTimer = null;
 37  this.upgradeTimeoutTimer = null;
 38  this.pingTimeoutTimer = null;
 39
 40  this.setTransport(transport);
 41  this.onOpen();
 42}
 43
 44/**
 45 * Inherits from EventEmitter.
 46 */
 47
 48util.inherits(Socket, EventEmitter);
 49
 50/**
 51 * Called upon transport considered open.
 52 *
 53 * @api private
 54 */
 55
 56Socket.prototype.onOpen = function () {
 57  this.readyState = 'open';
 58
 59  // sends an `open` packet
 60  this.transport.sid = this.id;
 61  this.sendPacket('open', JSON.stringify({
 62    sid: this.id,
 63    upgrades: this.getAvailableUpgrades(),
 64    pingInterval: this.server.pingInterval,
 65    pingTimeout: this.server.pingTimeout
 66  }));
 67
 68  this.emit('open');
 69  this.setPingTimeout();
 70};
 71
 72/**
 73 * Called upon transport packet.
 74 *
 75 * @param {Object} packet
 76 * @api private
 77 */
 78
 79Socket.prototype.onPacket = function (packet) {
 80  if ('open' === this.readyState) {
 81    // export packet event
 82    debug('packet');
 83    this.emit('packet', packet);
 84
 85    // Reset ping timeout on any packet, incoming data is a good sign of
 86    // other side's liveness
 87    this.setPingTimeout();
 88
 89    switch (packet.type) {
 90
 91      case 'ping':
 92        debug('got ping');
 93        this.sendPacket('pong');
 94        this.emit('heartbeat');
 95        break;
 96
 97      case 'error':
 98        this.onClose('parse error');
 99        break;
100
101      case 'message':
102        this.emit('data', packet.data);
103        this.emit('message', packet.data);
104        break;
105    }
106  } else {
107    debug('packet received with closed socket');
108  }
109};
110
111/**
112 * Called upon transport error.
113 *
114 * @param {Error} error object
115 * @api private
116 */
117
118Socket.prototype.onError = function (err) {
119  debug('transport error');
120  this.onClose('transport error', err);
121};
122
123/**
124 * Sets and resets ping timeout timer based on client pings.
125 *
126 * @api private
127 */
128
129Socket.prototype.setPingTimeout = function () {
130  var self = this;
131  clearTimeout(self.pingTimeoutTimer);
132  self.pingTimeoutTimer = setTimeout(function () {
133    self.onClose('ping timeout');
134  }, self.server.pingInterval + self.server.pingTimeout);
135};
136
137/**
138 * Attaches handlers for the given transport.
139 *
140 * @param {Transport} transport
141 * @api private
142 */
143
144Socket.prototype.setTransport = function (transport) {
145  var onError = this.onError.bind(this);
146  var onPacket = this.onPacket.bind(this);
147  var flush = this.flush.bind(this);
148  var onClose = this.onClose.bind(this, 'transport close');
149
150  this.transport = transport;
151  this.transport.once('error', onError);
152  this.transport.on('packet', onPacket);
153  this.transport.on('drain', flush);
154  this.transport.once('close', onClose);
155  // this function will manage packet events (also message callbacks)
156  this.setupSendCallback();
157
158  this.cleanupFn.push(function () {
159    transport.removeListener('error', onError);
160    transport.removeListener('packet', onPacket);
161    transport.removeListener('drain', flush);
162    transport.removeListener('close', onClose);
163  });
164};
165
166/**
167 * Upgrades socket to the given transport
168 *
169 * @param {Transport} transport
170 * @api private
171 */
172
173Socket.prototype.maybeUpgrade = function (transport) {
174  debug('might upgrade socket transport from "%s" to "%s"'
175    , this.transport.name, transport.name);
176
177  this.upgrading = true;
178
179  var self = this;
180
181  // set transport upgrade timer
182  self.upgradeTimeoutTimer = setTimeout(function () {
183    debug('client did not complete upgrade - closing transport');
184    cleanup();
185    if ('open' === transport.readyState) {
186      transport.close();
187    }
188  }, this.server.upgradeTimeout);
189
190  function onPacket (packet) {
191    if ('ping' === packet.type && 'probe' === packet.data) {
192      transport.send([{ type: 'pong', data: 'probe' }]);
193      self.emit('upgrading', transport);
194      clearInterval(self.checkIntervalTimer);
195      self.checkIntervalTimer = setInterval(check, 100);
196    } else if ('upgrade' === packet.type && self.readyState !== 'closed') {
197      debug('got upgrade packet - upgrading');
198      cleanup();
199      self.transport.discard();
200      self.upgraded = true;
201      self.clearTransport();
202      self.setTransport(transport);
203      self.emit('upgrade', transport);
204      self.setPingTimeout();
205      self.flush();
206      if (self.readyState === 'closing') {
207        transport.close(function () {
208          self.onClose('forced close');
209        });
210      }
211    } else {
212      cleanup();
213      transport.close();
214    }
215  }
216
217  // we force a polling cycle to ensure a fast upgrade
218  function check () {
219    if ('polling' === self.transport.name && self.transport.writable) {
220      debug('writing a noop packet to polling for fast upgrade');
221      self.transport.send([{ type: 'noop' }]);
222    }
223  }
224
225  function cleanup () {
226    self.upgrading = false;
227
228    clearInterval(self.checkIntervalTimer);
229    self.checkIntervalTimer = null;
230
231    clearTimeout(self.upgradeTimeoutTimer);
232    self.upgradeTimeoutTimer = null;
233
234    transport.removeListener('packet', onPacket);
235    transport.removeListener('close', onTransportClose);
236    transport.removeListener('error', onError);
237    self.removeListener('close', onClose);
238  }
239
240  function onError (err) {
241    debug('client did not complete upgrade - %s', err);
242    cleanup();
243    transport.close();
244    transport = null;
245  }
246
247  function onTransportClose () {
248    onError('transport closed');
249  }
250
251  function onClose () {
252    onError('socket closed');
253  }
254
255  transport.on('packet', onPacket);
256  transport.once('close', onTransportClose);
257  transport.once('error', onError);
258
259  self.once('close', onClose);
260};
261
262/**
263 * Clears listeners and timers associated with current transport.
264 *
265 * @api private
266 */
267
268Socket.prototype.clearTransport = function () {
269  var cleanup;
270
271  var toCleanUp = this.cleanupFn.length;
272
273  for (var i = 0; i < toCleanUp; i++) {
274    cleanup = this.cleanupFn.shift();
275    cleanup();
276  }
277
278  // silence further transport errors and prevent uncaught exceptions
279  this.transport.on('error', function () {
280    debug('error triggered by discarded transport');
281  });
282
283  // ensure transport won't stay open
284  this.transport.close();
285
286  clearTimeout(this.pingTimeoutTimer);
287};
288
289/**
290 * Called upon transport considered closed.
291 * Possible reasons: `ping timeout`, `client error`, `parse error`,
292 * `transport error`, `server close`, `transport close`
293 */
294
295Socket.prototype.onClose = function (reason, description) {
296  if ('closed' !== this.readyState) {
297    this.readyState = 'closed';
298    clearTimeout(this.pingTimeoutTimer);
299    clearInterval(this.checkIntervalTimer);
300    this.checkIntervalTimer = null;
301    clearTimeout(this.upgradeTimeoutTimer);
302    var self = this;
303    // clean writeBuffer in next tick, so developers can still
304    // grab the writeBuffer on 'close' event
305    process.nextTick(function () {
306      self.writeBuffer = [];
307    });
308    this.packetsFn = [];
309    this.sentCallbackFn = [];
310    this.clearTransport();
311    this.emit('close', reason, description);
312  }
313};
314
315/**
316 * Setup and manage send callback
317 *
318 * @api private
319 */
320
321Socket.prototype.setupSendCallback = function () {
322  var self = this;
323  this.transport.on('drain', onDrain);
324
325  this.cleanupFn.push(function () {
326    self.transport.removeListener('drain', onDrain);
327  });
328
329  // the message was sent successfully, execute the callback
330  function onDrain () {
331    if (self.sentCallbackFn.length > 0) {
332      var seqFn = self.sentCallbackFn.splice(0, 1)[0];
333      if ('function' === typeof seqFn) {
334        debug('executing send callback');
335        seqFn(self.transport);
336      } else if (Array.isArray(seqFn)) {
337        debug('executing batch send callback');
338        for (var l = seqFn.length, i = 0; i < l; i++) {
339          if ('function' === typeof seqFn[i]) {
340            seqFn[i](self.transport);
341          }
342        }
343      }
344    }
345  }
346};
347
348/**
349 * Sends a message packet.
350 *
351 * @param {String} message
352 * @param {Object} options
353 * @param {Function} callback
354 * @return {Socket} for chaining
355 * @api public
356 */
357
358Socket.prototype.send =
359Socket.prototype.write = function (data, options, callback) {
360  this.sendPacket('message', data, options, callback);
361  return this;
362};
363
364/**
365 * Sends a packet.
366 *
367 * @param {String} packet type
368 * @param {String} optional, data
369 * @param {Object} options
370 * @api private
371 */
372
373Socket.prototype.sendPacket = function (type, data, options, callback) {
374  if ('function' === typeof options) {
375    callback = options;
376    options = null;
377  }
378
379  options = options || {};
380  options.compress = false !== options.compress;
381
382  if ('closing' !== this.readyState) {
383    debug('sending packet "%s" (%s)', type, data);
384
385    var packet = {
386      type: type,
387      options: options
388    };
389    if (data) packet.data = data;
390
391    // exports packetCreate event
392    this.emit('packetCreate', packet);
393
394    this.writeBuffer.push(packet);
395
396    // add send callback to object, if defined
397    if (callback) this.packetsFn.push(callback);
398
399    this.flush();
400  }
401};
402
403/**
404 * Attempts to flush the packets buffer.
405 *
406 * @api private
407 */
408
409Socket.prototype.flush = function () {
410  if ('closed' !== this.readyState &&
411                this.transport.writable &&
412                this.writeBuffer.length) {
413    debug('flushing buffer to transport');
414    this.emit('flush', this.writeBuffer);
415    this.server.emit('flush', this, this.writeBuffer);
416    var wbuf = this.writeBuffer;
417    this.writeBuffer = [];
418    if (!this.transport.supportsFraming) {
419      this.sentCallbackFn.push(this.packetsFn);
420    } else {
421      this.sentCallbackFn.push.apply(this.sentCallbackFn, this.packetsFn);
422    }
423    this.packetsFn = [];
424    this.transport.send(wbuf);
425    this.emit('drain');
426    this.server.emit('drain', this);
427  }
428};
429
430/**
431 * Get available upgrades for this socket.
432 *
433 * @api private
434 */
435
436Socket.prototype.getAvailableUpgrades = function () {
437  var availableUpgrades = [];
438  var allUpgrades = this.server.upgrades(this.transport.name);
439  for (var i = 0, l = allUpgrades.length; i < l; ++i) {
440    var upg = allUpgrades[i];
441    if (this.server.transports.indexOf(upg) !== -1) {
442      availableUpgrades.push(upg);
443    }
444  }
445  return availableUpgrades;
446};
447
448/**
449 * Closes the socket and underlying transport.
450 *
451 * @param {Boolean} optional, discard
452 * @return {Socket} for chaining
453 * @api public
454 */
455
456Socket.prototype.close = function (discard) {
457  if ('open' !== this.readyState) return;
458
459  this.readyState = 'closing';
460
461  if (this.writeBuffer.length) {
462    this.once('drain', this.closeTransport.bind(this, discard));
463    return;
464  }
465
466  this.closeTransport(discard);
467};
468
469/**
470 * Closes the underlying transport.
471 *
472 * @param {Boolean} discard
473 * @api private
474 */
475
476Socket.prototype.closeTransport = function (discard) {
477  if (discard) this.transport.discard();
478  this.transport.close(this.onClose.bind(this, 'forced close'));
479};