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