1
  2/**
  3 * Module dependencies.
  4 */
  5
  6var eio = require('engine.io-client');
  7var Socket = require('./socket');
  8var Emitter = require('component-emitter');
  9var parser = require('socket.io-parser');
 10var on = require('./on');
 11var bind = require('component-bind');
 12var debug = require('debug')('socket.io-client:manager');
 13var indexOf = require('indexof');
 14var Backoff = require('backo2');
 15
 16/**
 17 * IE6+ hasOwnProperty
 18 */
 19
 20var has = Object.prototype.hasOwnProperty;
 21
 22/**
 23 * Module exports
 24 */
 25
 26module.exports = Manager;
 27
 28/**
 29 * `Manager` constructor.
 30 *
 31 * @param {String} engine instance or engine uri/opts
 32 * @param {Object} options
 33 * @api public
 34 */
 35
 36function Manager (uri, opts) {
 37  if (!(this instanceof Manager)) return new Manager(uri, opts);
 38  if (uri && ('object' === typeof uri)) {
 39    opts = uri;
 40    uri = undefined;
 41  }
 42  opts = opts || {};
 43
 44  opts.path = opts.path || '/socket.io';
 45  this.nsps = {};
 46  this.subs = [];
 47  this.opts = opts;
 48  this.reconnection(opts.reconnection !== false);
 49  this.reconnectionAttempts(opts.reconnectionAttempts || Infinity);
 50  this.reconnectionDelay(opts.reconnectionDelay || 1000);
 51  this.reconnectionDelayMax(opts.reconnectionDelayMax || 5000);
 52  this.randomizationFactor(opts.randomizationFactor || 0.5);
 53  this.backoff = new Backoff({
 54    min: this.reconnectionDelay(),
 55    max: this.reconnectionDelayMax(),
 56    jitter: this.randomizationFactor()
 57  });
 58  this.timeout(null == opts.timeout ? 20000 : opts.timeout);
 59  this.readyState = 'closed';
 60  this.uri = uri;
 61  this.connecting = [];
 62  this.lastPing = null;
 63  this.encoding = false;
 64  this.packetBuffer = [];
 65  this.encoder = new parser.Encoder();
 66  this.decoder = new parser.Decoder();
 67  this.autoConnect = opts.autoConnect !== false;
 68  if (this.autoConnect) this.open();
 69}
 70
 71/**
 72 * Propagate given event to sockets and emit on `this`
 73 *
 74 * @api private
 75 */
 76
 77Manager.prototype.emitAll = function () {
 78  this.emit.apply(this, arguments);
 79  for (var nsp in this.nsps) {
 80    if (has.call(this.nsps, nsp)) {
 81      this.nsps[nsp].emit.apply(this.nsps[nsp], arguments);
 82    }
 83  }
 84};
 85
 86/**
 87 * Update `socket.id` of all sockets
 88 *
 89 * @api private
 90 */
 91
 92Manager.prototype.updateSocketIds = function () {
 93  for (var nsp in this.nsps) {
 94    if (has.call(this.nsps, nsp)) {
 95      this.nsps[nsp].id = this.engine.id;
 96    }
 97  }
 98};
 99
100/**
101 * Mix in `Emitter`.
102 */
103
104Emitter(Manager.prototype);
105
106/**
107 * Sets the `reconnection` config.
108 *
109 * @param {Boolean} true/false if it should automatically reconnect
110 * @return {Manager} self or value
111 * @api public
112 */
113
114Manager.prototype.reconnection = function (v) {
115  if (!arguments.length) return this._reconnection;
116  this._reconnection = !!v;
117  return this;
118};
119
120/**
121 * Sets the reconnection attempts config.
122 *
123 * @param {Number} max reconnection attempts before giving up
124 * @return {Manager} self or value
125 * @api public
126 */
127
128Manager.prototype.reconnectionAttempts = function (v) {
129  if (!arguments.length) return this._reconnectionAttempts;
130  this._reconnectionAttempts = v;
131  return this;
132};
133
134/**
135 * Sets the delay between reconnections.
136 *
137 * @param {Number} delay
138 * @return {Manager} self or value
139 * @api public
140 */
141
142Manager.prototype.reconnectionDelay = function (v) {
143  if (!arguments.length) return this._reconnectionDelay;
144  this._reconnectionDelay = v;
145  this.backoff && this.backoff.setMin(v);
146  return this;
147};
148
149Manager.prototype.randomizationFactor = function (v) {
150  if (!arguments.length) return this._randomizationFactor;
151  this._randomizationFactor = v;
152  this.backoff && this.backoff.setJitter(v);
153  return this;
154};
155
156/**
157 * Sets the maximum delay between reconnections.
158 *
159 * @param {Number} delay
160 * @return {Manager} self or value
161 * @api public
162 */
163
164Manager.prototype.reconnectionDelayMax = function (v) {
165  if (!arguments.length) return this._reconnectionDelayMax;
166  this._reconnectionDelayMax = v;
167  this.backoff && this.backoff.setMax(v);
168  return this;
169};
170
171/**
172 * Sets the connection timeout. `false` to disable
173 *
174 * @return {Manager} self or value
175 * @api public
176 */
177
178Manager.prototype.timeout = function (v) {
179  if (!arguments.length) return this._timeout;
180  this._timeout = v;
181  return this;
182};
183
184/**
185 * Starts trying to reconnect if reconnection is enabled and we have not
186 * started reconnecting yet
187 *
188 * @api private
189 */
190
191Manager.prototype.maybeReconnectOnOpen = function () {
192  // Only try to reconnect if it's the first time we're connecting
193  if (!this.reconnecting && this._reconnection && this.backoff.attempts === 0) {
194    // keeps reconnection from firing twice for the same reconnection loop
195    this.reconnect();
196  }
197};
198
199/**
200 * Sets the current transport `socket`.
201 *
202 * @param {Function} optional, callback
203 * @return {Manager} self
204 * @api public
205 */
206
207Manager.prototype.open =
208Manager.prototype.connect = function (fn, opts) {
209  debug('readyState %s', this.readyState);
210  if (~this.readyState.indexOf('open')) return this;
211
212  debug('opening %s', this.uri);
213  this.engine = eio(this.uri, this.opts);
214  var socket = this.engine;
215  var self = this;
216  this.readyState = 'opening';
217  this.skipReconnect = false;
218
219  // emit `open`
220  var openSub = on(socket, 'open', function () {
221    self.onopen();
222    fn && fn();
223  });
224
225  // emit `connect_error`
226  var errorSub = on(socket, 'error', function (data) {
227    debug('connect_error');
228    self.cleanup();
229    self.readyState = 'closed';
230    self.emitAll('connect_error', data);
231    if (fn) {
232      var err = new Error('Connection error');
233      err.data = data;
234      fn(err);
235    } else {
236      // Only do this if there is no fn to handle the error
237      self.maybeReconnectOnOpen();
238    }
239  });
240
241  // emit `connect_timeout`
242  if (false !== this._timeout) {
243    var timeout = this._timeout;
244    debug('connect attempt will timeout after %d', timeout);
245
246    // set timer
247    var timer = setTimeout(function () {
248      debug('connect attempt timed out after %d', timeout);
249      openSub.destroy();
250      socket.close();
251      socket.emit('error', 'timeout');
252      self.emitAll('connect_timeout', timeout);
253    }, timeout);
254
255    this.subs.push({
256      destroy: function () {
257        clearTimeout(timer);
258      }
259    });
260  }
261
262  this.subs.push(openSub);
263  this.subs.push(errorSub);
264
265  return this;
266};
267
268/**
269 * Called upon transport open.
270 *
271 * @api private
272 */
273
274Manager.prototype.onopen = function () {
275  debug('open');
276
277  // clear old subs
278  this.cleanup();
279
280  // mark as open
281  this.readyState = 'open';
282  this.emit('open');
283
284  // add new subs
285  var socket = this.engine;
286  this.subs.push(on(socket, 'data', bind(this, 'ondata')));
287  this.subs.push(on(socket, 'ping', bind(this, 'onping')));
288  this.subs.push(on(socket, 'pong', bind(this, 'onpong')));
289  this.subs.push(on(socket, 'error', bind(this, 'onerror')));
290  this.subs.push(on(socket, 'close', bind(this, 'onclose')));
291  this.subs.push(on(this.decoder, 'decoded', bind(this, 'ondecoded')));
292};
293
294/**
295 * Called upon a ping.
296 *
297 * @api private
298 */
299
300Manager.prototype.onping = function () {
301  this.lastPing = new Date();
302  this.emitAll('ping');
303};
304
305/**
306 * Called upon a packet.
307 *
308 * @api private
309 */
310
311Manager.prototype.onpong = function () {
312  this.emitAll('pong', new Date() - this.lastPing);
313};
314
315/**
316 * Called with data.
317 *
318 * @api private
319 */
320
321Manager.prototype.ondata = function (data) {
322  this.decoder.add(data);
323};
324
325/**
326 * Called when parser fully decodes a packet.
327 *
328 * @api private
329 */
330
331Manager.prototype.ondecoded = function (packet) {
332  this.emit('packet', packet);
333};
334
335/**
336 * Called upon socket error.
337 *
338 * @api private
339 */
340
341Manager.prototype.onerror = function (err) {
342  debug('error', err);
343  this.emitAll('error', err);
344};
345
346/**
347 * Creates a new socket for the given `nsp`.
348 *
349 * @return {Socket}
350 * @api public
351 */
352
353Manager.prototype.socket = function (nsp, opts) {
354  var socket = this.nsps[nsp];
355  if (!socket) {
356    socket = new Socket(this, nsp, opts);
357    this.nsps[nsp] = socket;
358    var self = this;
359    socket.on('connecting', onConnecting);
360    socket.on('connect', function () {
361      socket.id = self.engine.id;
362    });
363
364    if (this.autoConnect) {
365      // manually call here since connecting evnet is fired before listening
366      onConnecting();
367    }
368  }
369
370  function onConnecting () {
371    if (!~indexOf(self.connecting, socket)) {
372      self.connecting.push(socket);
373    }
374  }
375
376  return socket;
377};
378
379/**
380 * Called upon a socket close.
381 *
382 * @param {Socket} socket
383 */
384
385Manager.prototype.destroy = function (socket) {
386  var index = indexOf(this.connecting, socket);
387  if (~index) this.connecting.splice(index, 1);
388  if (this.connecting.length) return;
389
390  this.close();
391};
392
393/**
394 * Writes a packet.
395 *
396 * @param {Object} packet
397 * @api private
398 */
399
400Manager.prototype.packet = function (packet) {
401  debug('writing packet %j', packet);
402  var self = this;
403  if (packet.query && packet.type === 0) packet.nsp += '?' + packet.query;
404
405  if (!self.encoding) {
406    // encode, then write to engine with result
407    self.encoding = true;
408    this.encoder.encode(packet, function (encodedPackets) {
409      for (var i = 0; i < encodedPackets.length; i++) {
410        self.engine.write(encodedPackets[i], packet.options);
411      }
412      self.encoding = false;
413      self.processPacketQueue();
414    });
415  } else { // add packet to the queue
416    self.packetBuffer.push(packet);
417  }
418};
419
420/**
421 * If packet buffer is non-empty, begins encoding the
422 * next packet in line.
423 *
424 * @api private
425 */
426
427Manager.prototype.processPacketQueue = function () {
428  if (this.packetBuffer.length > 0 && !this.encoding) {
429    var pack = this.packetBuffer.shift();
430    this.packet(pack);
431  }
432};
433
434/**
435 * Clean up transport subscriptions and packet buffer.
436 *
437 * @api private
438 */
439
440Manager.prototype.cleanup = function () {
441  debug('cleanup');
442
443  var subsLength = this.subs.length;
444  for (var i = 0; i < subsLength; i++) {
445    var sub = this.subs.shift();
446    sub.destroy();
447  }
448
449  this.packetBuffer = [];
450  this.encoding = false;
451  this.lastPing = null;
452
453  this.decoder.destroy();
454};
455
456/**
457 * Close the current socket.
458 *
459 * @api private
460 */
461
462Manager.prototype.close =
463Manager.prototype.disconnect = function () {
464  debug('disconnect');
465  this.skipReconnect = true;
466  this.reconnecting = false;
467  if ('opening' === this.readyState) {
468    // `onclose` will not fire because
469    // an open event never happened
470    this.cleanup();
471  }
472  this.backoff.reset();
473  this.readyState = 'closed';
474  if (this.engine) this.engine.close();
475};
476
477/**
478 * Called upon engine close.
479 *
480 * @api private
481 */
482
483Manager.prototype.onclose = function (reason) {
484  debug('onclose');
485
486  this.cleanup();
487  this.backoff.reset();
488  this.readyState = 'closed';
489  this.emit('close', reason);
490
491  if (this._reconnection && !this.skipReconnect) {
492    this.reconnect();
493  }
494};
495
496/**
497 * Attempt a reconnection.
498 *
499 * @api private
500 */
501
502Manager.prototype.reconnect = function () {
503  if (this.reconnecting || this.skipReconnect) return this;
504
505  var self = this;
506
507  if (this.backoff.attempts >= this._reconnectionAttempts) {
508    debug('reconnect failed');
509    this.backoff.reset();
510    this.emitAll('reconnect_failed');
511    this.reconnecting = false;
512  } else {
513    var delay = this.backoff.duration();
514    debug('will wait %dms before reconnect attempt', delay);
515
516    this.reconnecting = true;
517    var timer = setTimeout(function () {
518      if (self.skipReconnect) return;
519
520      debug('attempting reconnect');
521      self.emitAll('reconnect_attempt', self.backoff.attempts);
522      self.emitAll('reconnecting', self.backoff.attempts);
523
524      // check again for the case socket closed in above events
525      if (self.skipReconnect) return;
526
527      self.open(function (err) {
528        if (err) {
529          debug('reconnect attempt error');
530          self.reconnecting = false;
531          self.reconnect();
532          self.emitAll('reconnect_error', err.data);
533        } else {
534          debug('reconnect success');
535          self.onreconnect();
536        }
537      });
538    }, delay);
539
540    this.subs.push({
541      destroy: function () {
542        clearTimeout(timer);
543      }
544    });
545  }
546};
547
548/**
549 * Called upon successful reconnect.
550 *
551 * @api private
552 */
553
554Manager.prototype.onreconnect = function () {
555  var attempt = this.backoff.attempts;
556  this.reconnecting = false;
557  this.backoff.reset();
558  this.updateSocketIds();
559  this.emitAll('reconnect', attempt);
560};