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