1/**
  2 * Module dependencies.
  3 */
  4
  5var transports = require('./transports/index');
  6var Emitter = require('component-emitter');
  7var debug = require('debug')('engine.io-client:socket');
  8var index = require('indexof');
  9var parser = require('engine.io-parser');
 10var parseuri = require('parseuri');
 11var parsejson = require('parsejson');
 12var parseqs = require('parseqs');
 13
 14/**
 15 * Module exports.
 16 */
 17
 18module.exports = Socket;
 19
 20/**
 21 * Socket constructor.
 22 *
 23 * @param {String|Object} uri or options
 24 * @param {Object} options
 25 * @api public
 26 */
 27
 28function Socket (uri, opts) {
 29  if (!(this instanceof Socket)) return new Socket(uri, opts);
 30
 31  opts = opts || {};
 32
 33  if (uri && 'object' === typeof uri) {
 34    opts = uri;
 35    uri = null;
 36  }
 37
 38  if (uri) {
 39    uri = parseuri(uri);
 40    opts.hostname = uri.host;
 41    opts.secure = uri.protocol === 'https' || uri.protocol === 'wss';
 42    opts.port = uri.port;
 43    if (uri.query) opts.query = uri.query;
 44  } else if (opts.host) {
 45    opts.hostname = parseuri(opts.host).host;
 46  }
 47
 48  this.secure = null != opts.secure ? opts.secure
 49    : (global.location && 'https:' === location.protocol);
 50
 51  if (opts.hostname && !opts.port) {
 52    // if no port is specified manually, use the protocol default
 53    opts.port = this.secure ? '443' : '80';
 54  }
 55
 56  this.agent = opts.agent || false;
 57  this.hostname = opts.hostname ||
 58    (global.location ? location.hostname : 'localhost');
 59  this.port = opts.port || (global.location && location.port
 60      ? location.port
 61      : (this.secure ? 443 : 80));
 62  this.query = opts.query || {};
 63  if ('string' === typeof this.query) this.query = parseqs.decode(this.query);
 64  this.upgrade = false !== opts.upgrade;
 65  this.path = (opts.path || '/engine.io').replace(/\/$/, '') + '/';
 66  this.forceJSONP = !!opts.forceJSONP;
 67  this.jsonp = false !== opts.jsonp;
 68  this.forceBase64 = !!opts.forceBase64;
 69  this.enablesXDR = !!opts.enablesXDR;
 70  this.timestampParam = opts.timestampParam || 't';
 71  this.timestampRequests = opts.timestampRequests;
 72  this.transports = opts.transports || ['polling', 'websocket'];
 73  this.readyState = '';
 74  this.writeBuffer = [];
 75  this.prevBufferLen = 0;
 76  this.policyPort = opts.policyPort || 843;
 77  this.rememberUpgrade = opts.rememberUpgrade || false;
 78  this.binaryType = null;
 79  this.onlyBinaryUpgrades = opts.onlyBinaryUpgrades;
 80  this.perMessageDeflate = false !== opts.perMessageDeflate ? (opts.perMessageDeflate || {}) : false;
 81
 82  if (true === this.perMessageDeflate) this.perMessageDeflate = {};
 83  if (this.perMessageDeflate && null == this.perMessageDeflate.threshold) {
 84    this.perMessageDeflate.threshold = 1024;
 85  }
 86
 87  // SSL options for Node.js client
 88  this.pfx = opts.pfx || null;
 89  this.key = opts.key || null;
 90  this.passphrase = opts.passphrase || null;
 91  this.cert = opts.cert || null;
 92  this.ca = opts.ca || null;
 93  this.ciphers = opts.ciphers || null;
 94  this.rejectUnauthorized = opts.rejectUnauthorized === undefined ? null : opts.rejectUnauthorized;
 95  this.forceNode = !!opts.forceNode;
 96
 97  // other options for Node.js client
 98  var freeGlobal = typeof global === 'object' && global;
 99  if (freeGlobal.global === freeGlobal) {
100    if (opts.extraHeaders && Object.keys(opts.extraHeaders).length > 0) {
101      this.extraHeaders = opts.extraHeaders;
102    }
103
104    if (opts.localAddress) {
105      this.localAddress = opts.localAddress;
106    }
107  }
108
109  // set on handshake
110  this.id = null;
111  this.upgrades = null;
112  this.pingInterval = null;
113  this.pingTimeout = null;
114
115  // set on heartbeat
116  this.pingIntervalTimer = null;
117  this.pingTimeoutTimer = null;
118
119  this.open();
120}
121
122Socket.priorWebsocketSuccess = false;
123
124/**
125 * Mix in `Emitter`.
126 */
127
128Emitter(Socket.prototype);
129
130/**
131 * Protocol version.
132 *
133 * @api public
134 */
135
136Socket.protocol = parser.protocol; // this is an int
137
138/**
139 * Expose deps for legacy compatibility
140 * and standalone browser access.
141 */
142
143Socket.Socket = Socket;
144Socket.Transport = require('./transport');
145Socket.transports = require('./transports/index');
146Socket.parser = require('engine.io-parser');
147
148/**
149 * Creates transport of the given type.
150 *
151 * @param {String} transport name
152 * @return {Transport}
153 * @api private
154 */
155
156Socket.prototype.createTransport = function (name) {
157  debug('creating transport "%s"', name);
158  var query = clone(this.query);
159
160  // append engine.io protocol identifier
161  query.EIO = parser.protocol;
162
163  // transport name
164  query.transport = name;
165
166  // session id if we already have one
167  if (this.id) query.sid = this.id;
168
169  var transport = new transports[name]({
170    agent: this.agent,
171    hostname: this.hostname,
172    port: this.port,
173    secure: this.secure,
174    path: this.path,
175    query: query,
176    forceJSONP: this.forceJSONP,
177    jsonp: this.jsonp,
178    forceBase64: this.forceBase64,
179    enablesXDR: this.enablesXDR,
180    timestampRequests: this.timestampRequests,
181    timestampParam: this.timestampParam,
182    policyPort: this.policyPort,
183    socket: this,
184    pfx: this.pfx,
185    key: this.key,
186    passphrase: this.passphrase,
187    cert: this.cert,
188    ca: this.ca,
189    ciphers: this.ciphers,
190    rejectUnauthorized: this.rejectUnauthorized,
191    perMessageDeflate: this.perMessageDeflate,
192    extraHeaders: this.extraHeaders,
193    forceNode: this.forceNode,
194    localAddress: this.localAddress
195  });
196
197  return transport;
198};
199
200function clone (obj) {
201  var o = {};
202  for (var i in obj) {
203    if (obj.hasOwnProperty(i)) {
204      o[i] = obj[i];
205    }
206  }
207  return o;
208}
209
210/**
211 * Initializes transport to use and starts probe.
212 *
213 * @api private
214 */
215Socket.prototype.open = function () {
216  var transport;
217  if (this.rememberUpgrade && Socket.priorWebsocketSuccess && this.transports.indexOf('websocket') !== -1) {
218    transport = 'websocket';
219  } else if (0 === this.transports.length) {
220    // Emit error on next tick so it can be listened to
221    var self = this;
222    setTimeout(function () {
223      self.emit('error', 'No transports available');
224    }, 0);
225    return;
226  } else {
227    transport = this.transports[0];
228  }
229  this.readyState = 'opening';
230
231  // Retry with the next transport if the transport is disabled (jsonp: false)
232  try {
233    transport = this.createTransport(transport);
234  } catch (e) {
235    this.transports.shift();
236    this.open();
237    return;
238  }
239
240  transport.open();
241  this.setTransport(transport);
242};
243
244/**
245 * Sets the current transport. Disables the existing one (if any).
246 *
247 * @api private
248 */
249
250Socket.prototype.setTransport = function (transport) {
251  debug('setting transport %s', transport.name);
252  var self = this;
253
254  if (this.transport) {
255    debug('clearing existing transport %s', this.transport.name);
256    this.transport.removeAllListeners();
257  }
258
259  // set up transport
260  this.transport = transport;
261
262  // set up transport listeners
263  transport
264  .on('drain', function () {
265    self.onDrain();
266  })
267  .on('packet', function (packet) {
268    self.onPacket(packet);
269  })
270  .on('error', function (e) {
271    self.onError(e);
272  })
273  .on('close', function () {
274    self.onClose('transport close');
275  });
276};
277
278/**
279 * Probes a transport.
280 *
281 * @param {String} transport name
282 * @api private
283 */
284
285Socket.prototype.probe = function (name) {
286  debug('probing transport "%s"', name);
287  var transport = this.createTransport(name, { probe: 1 });
288  var failed = false;
289  var self = this;
290
291  Socket.priorWebsocketSuccess = false;
292
293  function onTransportOpen () {
294    if (self.onlyBinaryUpgrades) {
295      var upgradeLosesBinary = !this.supportsBinary && self.transport.supportsBinary;
296      failed = failed || upgradeLosesBinary;
297    }
298    if (failed) return;
299
300    debug('probe transport "%s" opened', name);
301    transport.send([{ type: 'ping', data: 'probe' }]);
302    transport.once('packet', function (msg) {
303      if (failed) return;
304      if ('pong' === msg.type && 'probe' === msg.data) {
305        debug('probe transport "%s" pong', name);
306        self.upgrading = true;
307        self.emit('upgrading', transport);
308        if (!transport) return;
309        Socket.priorWebsocketSuccess = 'websocket' === transport.name;
310
311        debug('pausing current transport "%s"', self.transport.name);
312        self.transport.pause(function () {
313          if (failed) return;
314          if ('closed' === self.readyState) return;
315          debug('changing transport and sending upgrade packet');
316
317          cleanup();
318
319          self.setTransport(transport);
320          transport.send([{ type: 'upgrade' }]);
321          self.emit('upgrade', transport);
322          transport = null;
323          self.upgrading = false;
324          self.flush();
325        });
326      } else {
327        debug('probe transport "%s" failed', name);
328        var err = new Error('probe error');
329        err.transport = transport.name;
330        self.emit('upgradeError', err);
331      }
332    });
333  }
334
335  function freezeTransport () {
336    if (failed) return;
337
338    // Any callback called by transport should be ignored since now
339    failed = true;
340
341    cleanup();
342
343    transport.close();
344    transport = null;
345  }
346
347  // Handle any error that happens while probing
348  function onerror (err) {
349    var error = new Error('probe error: ' + err);
350    error.transport = transport.name;
351
352    freezeTransport();
353
354    debug('probe transport "%s" failed because of error: %s', name, err);
355
356    self.emit('upgradeError', error);
357  }
358
359  function onTransportClose () {
360    onerror('transport closed');
361  }
362
363  // When the socket is closed while we're probing
364  function onclose () {
365    onerror('socket closed');
366  }
367
368  // When the socket is upgraded while we're probing
369  function onupgrade (to) {
370    if (transport && to.name !== transport.name) {
371      debug('"%s" works - aborting "%s"', to.name, transport.name);
372      freezeTransport();
373    }
374  }
375
376  // Remove all listeners on the transport and on self
377  function cleanup () {
378    transport.removeListener('open', onTransportOpen);
379    transport.removeListener('error', onerror);
380    transport.removeListener('close', onTransportClose);
381    self.removeListener('close', onclose);
382    self.removeListener('upgrading', onupgrade);
383  }
384
385  transport.once('open', onTransportOpen);
386  transport.once('error', onerror);
387  transport.once('close', onTransportClose);
388
389  this.once('close', onclose);
390  this.once('upgrading', onupgrade);
391
392  transport.open();
393};
394
395/**
396 * Called when connection is deemed open.
397 *
398 * @api public
399 */
400
401Socket.prototype.onOpen = function () {
402  debug('socket open');
403  this.readyState = 'open';
404  Socket.priorWebsocketSuccess = 'websocket' === this.transport.name;
405  this.emit('open');
406  this.flush();
407
408  // we check for `readyState` in case an `open`
409  // listener already closed the socket
410  if ('open' === this.readyState && this.upgrade && this.transport.pause) {
411    debug('starting upgrade probes');
412    for (var i = 0, l = this.upgrades.length; i < l; i++) {
413      this.probe(this.upgrades[i]);
414    }
415  }
416};
417
418/**
419 * Handles a packet.
420 *
421 * @api private
422 */
423
424Socket.prototype.onPacket = function (packet) {
425  if ('opening' === this.readyState || 'open' === this.readyState ||
426      'closing' === this.readyState) {
427    debug('socket receive: type "%s", data "%s"', packet.type, packet.data);
428
429    this.emit('packet', packet);
430
431    // Socket is live - any packet counts
432    this.emit('heartbeat');
433
434    switch (packet.type) {
435      case 'open':
436        this.onHandshake(parsejson(packet.data));
437        break;
438
439      case 'pong':
440        this.setPing();
441        this.emit('pong');
442        break;
443
444      case 'error':
445        var err = new Error('server error');
446        err.code = packet.data;
447        this.onError(err);
448        break;
449
450      case 'message':
451        this.emit('data', packet.data);
452        this.emit('message', packet.data);
453        break;
454    }
455  } else {
456    debug('packet received with socket readyState "%s"', this.readyState);
457  }
458};
459
460/**
461 * Called upon handshake completion.
462 *
463 * @param {Object} handshake obj
464 * @api private
465 */
466
467Socket.prototype.onHandshake = function (data) {
468  this.emit('handshake', data);
469  this.id = data.sid;
470  this.transport.query.sid = data.sid;
471  this.upgrades = this.filterUpgrades(data.upgrades);
472  this.pingInterval = data.pingInterval;
473  this.pingTimeout = data.pingTimeout;
474  this.onOpen();
475  // In case open handler closes socket
476  if ('closed' === this.readyState) return;
477  this.setPing();
478
479  // Prolong liveness of socket on heartbeat
480  this.removeListener('heartbeat', this.onHeartbeat);
481  this.on('heartbeat', this.onHeartbeat);
482};
483
484/**
485 * Resets ping timeout.
486 *
487 * @api private
488 */
489
490Socket.prototype.onHeartbeat = function (timeout) {
491  clearTimeout(this.pingTimeoutTimer);
492  var self = this;
493  self.pingTimeoutTimer = setTimeout(function () {
494    if ('closed' === self.readyState) return;
495    self.onClose('ping timeout');
496  }, timeout || (self.pingInterval + self.pingTimeout));
497};
498
499/**
500 * Pings server every `this.pingInterval` and expects response
501 * within `this.pingTimeout` or closes connection.
502 *
503 * @api private
504 */
505
506Socket.prototype.setPing = function () {
507  var self = this;
508  clearTimeout(self.pingIntervalTimer);
509  self.pingIntervalTimer = setTimeout(function () {
510    debug('writing ping packet - expecting pong within %sms', self.pingTimeout);
511    self.ping();
512    self.onHeartbeat(self.pingTimeout);
513  }, self.pingInterval);
514};
515
516/**
517* Sends a ping packet.
518*
519* @api private
520*/
521
522Socket.prototype.ping = function () {
523  var self = this;
524  this.sendPacket('ping', function () {
525    self.emit('ping');
526  });
527};
528
529/**
530 * Called on `drain` event
531 *
532 * @api private
533 */
534
535Socket.prototype.onDrain = function () {
536  this.writeBuffer.splice(0, this.prevBufferLen);
537
538  // setting prevBufferLen = 0 is very important
539  // for example, when upgrading, upgrade packet is sent over,
540  // and a nonzero prevBufferLen could cause problems on `drain`
541  this.prevBufferLen = 0;
542
543  if (0 === this.writeBuffer.length) {
544    this.emit('drain');
545  } else {
546    this.flush();
547  }
548};
549
550/**
551 * Flush write buffers.
552 *
553 * @api private
554 */
555
556Socket.prototype.flush = function () {
557  if ('closed' !== this.readyState && this.transport.writable &&
558    !this.upgrading && this.writeBuffer.length) {
559    debug('flushing %d packets in socket', this.writeBuffer.length);
560    this.transport.send(this.writeBuffer);
561    // keep track of current length of writeBuffer
562    // splice writeBuffer and callbackBuffer on `drain`
563    this.prevBufferLen = this.writeBuffer.length;
564    this.emit('flush');
565  }
566};
567
568/**
569 * Sends a message.
570 *
571 * @param {String} message.
572 * @param {Function} callback function.
573 * @param {Object} options.
574 * @return {Socket} for chaining.
575 * @api public
576 */
577
578Socket.prototype.write =
579Socket.prototype.send = function (msg, options, fn) {
580  this.sendPacket('message', msg, options, fn);
581  return this;
582};
583
584/**
585 * Sends a packet.
586 *
587 * @param {String} packet type.
588 * @param {String} data.
589 * @param {Object} options.
590 * @param {Function} callback function.
591 * @api private
592 */
593
594Socket.prototype.sendPacket = function (type, data, options, fn) {
595  if ('function' === typeof data) {
596    fn = data;
597    data = undefined;
598  }
599
600  if ('function' === typeof options) {
601    fn = options;
602    options = null;
603  }
604
605  if ('closing' === this.readyState || 'closed' === this.readyState) {
606    return;
607  }
608
609  options = options || {};
610  options.compress = false !== options.compress;
611
612  var packet = {
613    type: type,
614    data: data,
615    options: options
616  };
617  this.emit('packetCreate', packet);
618  this.writeBuffer.push(packet);
619  if (fn) this.once('flush', fn);
620  this.flush();
621};
622
623/**
624 * Closes the connection.
625 *
626 * @api private
627 */
628
629Socket.prototype.close = function () {
630  if ('opening' === this.readyState || 'open' === this.readyState) {
631    this.readyState = 'closing';
632
633    var self = this;
634
635    if (this.writeBuffer.length) {
636      this.once('drain', function () {
637        if (this.upgrading) {
638          waitForUpgrade();
639        } else {
640          close();
641        }
642      });
643    } else if (this.upgrading) {
644      waitForUpgrade();
645    } else {
646      close();
647    }
648  }
649
650  function close () {
651    self.onClose('forced close');
652    debug('socket closing - telling transport to close');
653    self.transport.close();
654  }
655
656  function cleanupAndClose () {
657    self.removeListener('upgrade', cleanupAndClose);
658    self.removeListener('upgradeError', cleanupAndClose);
659    close();
660  }
661
662  function waitForUpgrade () {
663    // wait for upgrade to finish since we can't send packets while pausing a transport
664    self.once('upgrade', cleanupAndClose);
665    self.once('upgradeError', cleanupAndClose);
666  }
667
668  return this;
669};
670
671/**
672 * Called upon transport error
673 *
674 * @api private
675 */
676
677Socket.prototype.onError = function (err) {
678  debug('socket error %j', err);
679  Socket.priorWebsocketSuccess = false;
680  this.emit('error', err);
681  this.onClose('transport error', err);
682};
683
684/**
685 * Called upon transport close.
686 *
687 * @api private
688 */
689
690Socket.prototype.onClose = function (reason, desc) {
691  if ('opening' === this.readyState || 'open' === this.readyState || 'closing' === this.readyState) {
692    debug('socket close with reason: "%s"', reason);
693    var self = this;
694
695    // clear timers
696    clearTimeout(this.pingIntervalTimer);
697    clearTimeout(this.pingTimeoutTimer);
698
699    // stop event from firing again for transport
700    this.transport.removeAllListeners('close');
701
702    // ensure transport won't stay open
703    this.transport.close();
704
705    // ignore further transport communication
706    this.transport.removeAllListeners();
707
708    // set ready state
709    this.readyState = 'closed';
710
711    // clear session id
712    this.id = null;
713
714    // emit close event
715    this.emit('close', reason, desc);
716
717    // clean buffers after, so users can still
718    // grab the buffers on `close` event
719    self.writeBuffer = [];
720    self.prevBufferLen = 0;
721  }
722};
723
724/**
725 * Filters upgrades, returning only those matching client transports.
726 *
727 * @param {Array} server upgrades
728 * @api private
729 *
730 */
731
732Socket.prototype.filterUpgrades = function (upgrades) {
733  var filteredUpgrades = [];
734  for (var i = 0, j = upgrades.length; i < j; i++) {
735    if (~index(this.transports, upgrades[i])) filteredUpgrades.push(upgrades[i]);
736  }
737  return filteredUpgrades;
738};