|
|
/** * Module dependencies. */
var url = require('./url'); var eio = require('engine.io-client'); var Socket = require('./socket'); var Emitter = require('component-emitter'); var parser = require('socket.io-parser'); var on = require('./on'); var bind = require('component-bind'); var object = require('object-component'); var debug = require('debug')('socket.io-client:manager'); var indexOf = require('indexof');
/** * Module exports */
module.exports = Manager;
/** * `Manager` constructor. * * @param {String} engine instance or engine uri/opts * @param {Object} options * @api public */
function Manager(uri, opts){ if (!(this instanceof Manager)) return new Manager(uri, opts); if (uri && ('object' == typeof uri)) { opts = uri; uri = undefined; } opts = opts || {};
opts.path = opts.path || '/socket.io'; this.nsps = {}; this.subs = []; this.opts = opts; this.reconnection(opts.reconnection !== false); this.reconnectionAttempts(opts.reconnectionAttempts || Infinity); this.reconnectionDelay(opts.reconnectionDelay || 1000); this.reconnectionDelayMax(opts.reconnectionDelayMax || 5000); this.timeout(null == opts.timeout ? 20000 : opts.timeout); this.readyState = 'closed'; this.uri = uri; this.connected = []; this.attempts = 0; this.encoding = false; this.packetBuffer = []; this.encoder = new parser.Encoder(); this.decoder = new parser.Decoder(); this.autoConnect = opts.autoConnect !== false; if (this.autoConnect) this.open(); }
/** * Propagate given event to sockets and emit on `this` * * @api private */
Manager.prototype.emitAll = function() { this.emit.apply(this, arguments); for (var nsp in this.nsps) { this.nsps[nsp].emit.apply(this.nsps[nsp], arguments); } };
/** * Mix in `Emitter`. */
Emitter(Manager.prototype);
/** * Sets the `reconnection` config. * * @param {Boolean} true/false if it should automatically reconnect * @return {Manager} self or value * @api public */
Manager.prototype.reconnection = function(v){ if (!arguments.length) return this._reconnection; this._reconnection = !!v; return this; };
/** * Sets the reconnection attempts config. * * @param {Number} max reconnection attempts before giving up * @return {Manager} self or value * @api public */
Manager.prototype.reconnectionAttempts = function(v){ if (!arguments.length) return this._reconnectionAttempts; this._reconnectionAttempts = v; return this; };
/** * Sets the delay between reconnections. * * @param {Number} delay * @return {Manager} self or value * @api public */
Manager.prototype.reconnectionDelay = function(v){ if (!arguments.length) return this._reconnectionDelay; this._reconnectionDelay = v; return this; };
/** * Sets the maximum delay between reconnections. * * @param {Number} delay * @return {Manager} self or value * @api public */
Manager.prototype.reconnectionDelayMax = function(v){ if (!arguments.length) return this._reconnectionDelayMax; this._reconnectionDelayMax = v; return this; };
/** * Sets the connection timeout. `false` to disable * * @return {Manager} self or value * @api public */
Manager.prototype.timeout = function(v){ if (!arguments.length) return this._timeout; this._timeout = v; return this; };
/** * Starts trying to reconnect if reconnection is enabled and we have not * started reconnecting yet * * @api private */
Manager.prototype.maybeReconnectOnOpen = function() { // Only try to reconnect if it's the first time we're connecting
if (!this.openReconnect && !this.reconnecting && this._reconnection && this.attempts === 0) { // keeps reconnection from firing twice for the same reconnection loop
this.openReconnect = true; this.reconnect(); } };
/** * Sets the current transport `socket`. * * @param {Function} optional, callback * @return {Manager} self * @api public */
Manager.prototype.open = Manager.prototype.connect = function(fn){ debug('readyState %s', this.readyState); if (~this.readyState.indexOf('open')) return this;
debug('opening %s', this.uri); this.engine = eio(this.uri, this.opts); var socket = this.engine; var self = this; this.readyState = 'opening'; this.skipReconnect = false;
// emit `open`
var openSub = on(socket, 'open', function() { self.onopen(); fn && fn(); });
// emit `connect_error`
var errorSub = on(socket, 'error', function(data){ debug('connect_error'); self.cleanup(); self.readyState = 'closed'; self.emitAll('connect_error', data); if (fn) { var err = new Error('Connection error'); err.data = data; fn(err); }
self.maybeReconnectOnOpen(); });
// emit `connect_timeout`
if (false !== this._timeout) { var timeout = this._timeout; debug('connect attempt will timeout after %d', timeout);
// set timer
var timer = setTimeout(function(){ debug('connect attempt timed out after %d', timeout); openSub.destroy(); socket.close(); socket.emit('error', 'timeout'); self.emitAll('connect_timeout', timeout); }, timeout);
this.subs.push({ destroy: function(){ clearTimeout(timer); } }); }
this.subs.push(openSub); this.subs.push(errorSub);
return this; };
/** * Called upon transport open. * * @api private */
Manager.prototype.onopen = function(){ debug('open');
// clear old subs
this.cleanup();
// mark as open
this.readyState = 'open'; this.emit('open');
// add new subs
var socket = this.engine; this.subs.push(on(socket, 'data', bind(this, 'ondata'))); this.subs.push(on(this.decoder, 'decoded', bind(this, 'ondecoded'))); this.subs.push(on(socket, 'error', bind(this, 'onerror'))); this.subs.push(on(socket, 'close', bind(this, 'onclose'))); };
/** * Called with data. * * @api private */
Manager.prototype.ondata = function(data){ this.decoder.add(data); };
/** * Called when parser fully decodes a packet. * * @api private */
Manager.prototype.ondecoded = function(packet) { this.emit('packet', packet); };
/** * Called upon socket error. * * @api private */
Manager.prototype.onerror = function(err){ debug('error', err); this.emitAll('error', err); };
/** * Creates a new socket for the given `nsp`. * * @return {Socket} * @api public */
Manager.prototype.socket = function(nsp){ var socket = this.nsps[nsp]; if (!socket) { socket = new Socket(this, nsp); this.nsps[nsp] = socket; var self = this; socket.on('connect', function(){ if (!~indexOf(self.connected, socket)) { self.connected.push(socket); } }); } return socket; };
/** * Called upon a socket close. * * @param {Socket} socket */
Manager.prototype.destroy = function(socket){ var index = indexOf(this.connected, socket); if (~index) this.connected.splice(index, 1); if (this.connected.length) return;
this.close(); };
/** * Writes a packet. * * @param {Object} packet * @api private */
Manager.prototype.packet = function(packet){ debug('writing packet %j', packet); var self = this;
if (!self.encoding) { // encode, then write to engine with result
self.encoding = true; this.encoder.encode(packet, function(encodedPackets) { for (var i = 0; i < encodedPackets.length; i++) { self.engine.write(encodedPackets[i]); } self.encoding = false; self.processPacketQueue(); }); } else { // add packet to the queue
self.packetBuffer.push(packet); } };
/** * If packet buffer is non-empty, begins encoding the * next packet in line. * * @api private */
Manager.prototype.processPacketQueue = function() { if (this.packetBuffer.length > 0 && !this.encoding) { var pack = this.packetBuffer.shift(); this.packet(pack); } };
/** * Clean up transport subscriptions and packet buffer. * * @api private */
Manager.prototype.cleanup = function(){ var sub; while (sub = this.subs.shift()) sub.destroy();
this.packetBuffer = []; this.encoding = false;
this.decoder.destroy(); };
/** * Close the current socket. * * @api private */
Manager.prototype.close = Manager.prototype.disconnect = function(){ this.skipReconnect = true; this.readyState = 'closed'; this.engine && this.engine.close(); };
/** * Called upon engine close. * * @api private */
Manager.prototype.onclose = function(reason){ debug('close'); this.cleanup(); this.readyState = 'closed'; this.emit('close', reason); if (this._reconnection && !this.skipReconnect) { this.reconnect(); } };
/** * Attempt a reconnection. * * @api private */
Manager.prototype.reconnect = function(){ if (this.reconnecting || this.skipReconnect) return this;
var self = this; this.attempts++;
if (this.attempts > this._reconnectionAttempts) { debug('reconnect failed'); this.emitAll('reconnect_failed'); this.reconnecting = false; } else { var delay = this.attempts * this.reconnectionDelay(); delay = Math.min(delay, this.reconnectionDelayMax()); debug('will wait %dms before reconnect attempt', delay);
this.reconnecting = true; var timer = setTimeout(function(){ if (self.skipReconnect) return;
debug('attempting reconnect'); self.emitAll('reconnect_attempt', self.attempts); self.emitAll('reconnecting', self.attempts);
// check again for the case socket closed in above events
if (self.skipReconnect) return;
self.open(function(err){ if (err) { debug('reconnect attempt error'); self.reconnecting = false; self.reconnect(); self.emitAll('reconnect_error', err.data); } else { debug('reconnect success'); self.onreconnect(); } }); }, delay);
this.subs.push({ destroy: function(){ clearTimeout(timer); } }); } };
/** * Called upon successful reconnect. * * @api private */
Manager.prototype.onreconnect = function(){ var attempt = this.attempts; this.attempts = 0; this.reconnecting = false; this.emitAll('reconnect', attempt); };
|