1/**
  2 * Module dependencies.
  3 */
  4
  5var EventEmitter = require('events').EventEmitter;
  6var debug = require('debug')('engine:socket');
  7
  8/**
  9 * Module exports.
 10 */
 11
 12module.exports = Socket;
 13
 14/**
 15 * Client class (abstract).
 16 *
 17 * @api private
 18 */
 19
 20function Socket (id, server, transport, req) {
 21  this.id = id;
 22  this.server = server;
 23  this.upgraded = false;
 24  this.readyState = 'opening';
 25  this.writeBuffer = [];
 26  this.packetsFn = [];
 27  this.sentCallbackFn = [];
 28  this.request = req;
 29
 30  // Cache IP since it might not be in the req later
 31  this.remoteAddress = req.connection.remoteAddress;
 32
 33  this.checkIntervalTimer = null;
 34  this.upgradeTimeoutTimer = null;
 35  this.pingTimeoutTimer = null;
 36
 37  this.setTransport(transport);
 38  this.onOpen();
 39}
 40
 41/**
 42 * Inherits from EventEmitter.
 43 */
 44
 45Socket.prototype.__proto__ = EventEmitter.prototype;
 46
 47/**
 48 * Called upon transport considered open.
 49 *
 50 * @api private
 51 */
 52
 53Socket.prototype.onOpen = function () {
 54  this.readyState = 'open';
 55
 56  // sends an `open` packet
 57  this.transport.sid = this.id;
 58  this.sendPacket('open', JSON.stringify({
 59      sid: this.id
 60    , upgrades: this.getAvailableUpgrades()
 61    , pingInterval: this.server.pingInterval
 62    , pingTimeout: this.server.pingTimeout
 63  }));
 64
 65  this.emit('open');
 66  this.setPingTimeout();
 67};
 68
 69/**
 70 * Called upon transport packet.
 71 *
 72 * @param {Object} packet
 73 * @api private
 74 */
 75
 76Socket.prototype.onPacket = function (packet) {
 77  if ('open' == this.readyState) {
 78    // export packet event
 79    debug('packet');
 80    this.emit('packet', packet);
 81
 82    // Reset ping timeout on any packet, incoming data is a good sign of
 83    // other side's liveness
 84    this.setPingTimeout();
 85
 86    switch (packet.type) {
 87
 88      case 'ping':
 89        debug('got ping');
 90        this.sendPacket('pong');
 91        this.emit('heartbeat');
 92        break;
 93
 94      case 'error':
 95        this.onClose('parse error');
 96        break;
 97
 98      case 'message':
 99        this.emit('data', packet.data);
100        this.emit('message', packet.data);
101        break;
102    }
103  } else {
104    debug('packet received with closed socket');
105  }
106};
107
108/**
109 * Called upon transport error.
110 *
111 * @param {Error} error object
112 * @api private
113 */
114
115Socket.prototype.onError = function (err) {
116  debug('transport error');
117  this.onClose('transport error', err);
118};
119
120/**
121 * Sets and resets ping timeout timer based on client pings.
122 *
123 * @api private
124 */
125
126Socket.prototype.setPingTimeout = function () {
127  var self = this;
128  clearTimeout(self.pingTimeoutTimer);
129  self.pingTimeoutTimer = setTimeout(function () {
130    self.onClose('ping timeout');
131  }, self.server.pingInterval + self.server.pingTimeout);
132};
133
134/**
135 * Attaches handlers for the given transport.
136 *
137 * @param {Transport} transport
138 * @api private
139 */
140
141Socket.prototype.setTransport = function (transport) {
142  this.transport = transport;
143  this.transport.once('error', this.onError.bind(this));
144  this.transport.on('packet', this.onPacket.bind(this));
145  this.transport.on('drain', this.flush.bind(this));
146  this.transport.once('close', this.onClose.bind(this, 'transport close'));
147  //this function will manage packet events (also message callbacks)
148  this.setupSendCallback();
149};
150
151/**
152 * Upgrades socket to the given transport
153 *
154 * @param {Transport} transport
155 * @api private
156 */
157
158Socket.prototype.maybeUpgrade = function (transport) {
159  debug('might upgrade socket transport from "%s" to "%s"'
160    , this.transport.name, transport.name);
161
162  var self = this;
163
164  // set transport upgrade timer
165  self.upgradeTimeoutTimer = setTimeout(function () {
166    debug('client did not complete upgrade - closing transport');
167    clearInterval(self.checkIntervalTimer);
168    self.checkIntervalTimer = null;
169    if ('open' == transport.readyState) {
170      transport.close();
171    }
172  }, this.server.upgradeTimeout);
173
174  function onPacket(packet){
175    if ('ping' == packet.type && 'probe' == packet.data) {
176      transport.send([{ type: 'pong', data: 'probe' }]);
177      clearInterval(self.checkIntervalTimer);
178      self.checkIntervalTimer = setInterval(check, 100);
179    } else if ('upgrade' == packet.type && self.readyState != 'closed') {
180      debug('got upgrade packet - upgrading');
181      self.upgraded = true;
182      self.clearTransport();
183      self.setTransport(transport);
184      self.emit('upgrade', transport);
185      self.setPingTimeout();
186      self.flush();
187      clearInterval(self.checkIntervalTimer);
188      self.checkIntervalTimer = null;
189      clearTimeout(self.upgradeTimeoutTimer);
190      transport.removeListener('packet', onPacket);
191      if (self.readyState == 'closing') {
192        transport.close(function () {
193          self.onClose('forced close');
194        });
195      }
196    } else {
197      transport.close();
198    }
199  }
200
201  // we force a polling cycle to ensure a fast upgrade
202  function check(){
203    if ('polling' == self.transport.name && self.transport.writable) {
204      debug('writing a noop packet to polling for fast upgrade');
205      self.transport.send([{ type: 'noop' }]);
206    }
207  }
208
209  transport.on('packet', onPacket);
210};
211
212/**
213 * Clears listeners and timers associated with current transport.
214 *
215 * @api private
216 */
217
218Socket.prototype.clearTransport = function () {
219  // silence further transport errors and prevent uncaught exceptions
220  this.transport.on('error', function(){
221    debug('error triggered by discarded transport');
222  });
223  clearTimeout(this.pingTimeoutTimer);
224};
225
226/**
227 * Called upon transport considered closed.
228 * Possible reasons: `ping timeout`, `client error`, `parse error`,
229 * `transport error`, `server close`, `transport close`
230 */
231
232Socket.prototype.onClose = function (reason, description) {
233  if ('closed' != this.readyState) {
234    clearTimeout(this.pingTimeoutTimer);
235    clearInterval(this.checkIntervalTimer);
236    this.checkIntervalTimer = null;
237    clearTimeout(this.upgradeTimeoutTimer);
238    var self = this;
239    // clean writeBuffer in next tick, so developers can still
240    // grab the writeBuffer on 'close' event
241    process.nextTick(function() {
242      self.writeBuffer = [];
243    });
244    this.packetsFn = [];
245    this.sentCallbackFn = [];
246    this.clearTransport();
247    this.readyState = 'closed';
248    this.emit('close', reason, description);
249  }
250};
251
252/**
253 * Setup and manage send callback
254 *
255 * @api private
256 */
257
258Socket.prototype.setupSendCallback = function () {
259  var self = this;
260  //the message was sent successfully, execute the callback
261  this.transport.on('drain', function() {
262    if (self.sentCallbackFn.length > 0) {
263      var seqFn = self.sentCallbackFn.splice(0,1)[0];
264      if ('function' == typeof seqFn) {
265        debug('executing send callback');
266        seqFn(self.transport);
267      } else if (Array.isArray(seqFn)) {
268        debug('executing batch send callback');
269        for (var l = seqFn.length, i = 0; i < l; i++) {
270          if ('function' == typeof seqFn[i]) {
271            seqFn[i](self.transport);
272          }
273        }
274      }
275    }
276  });
277};
278
279/**
280 * Sends a message packet.
281 *
282 * @param {String} message
283 * @param {Function} callback
284 * @return {Socket} for chaining
285 * @api public
286 */
287
288Socket.prototype.send =
289Socket.prototype.write = function(data, callback){
290  this.sendPacket('message', data, callback);
291  return this;
292};
293
294/**
295 * Sends a packet.
296 *
297 * @param {String} packet type
298 * @param {String} optional, data
299 * @api private
300 */
301
302Socket.prototype.sendPacket = function (type, data, callback) {
303  if ('closing' != this.readyState) {
304    debug('sending packet "%s" (%s)', type, data);
305
306    var packet = { type: type };
307    if (data) packet.data = data;
308
309    // exports packetCreate event
310    this.emit('packetCreate', packet);
311
312    this.writeBuffer.push(packet);
313
314    //add send callback to object
315    this.packetsFn.push(callback);
316
317    this.flush();
318  }
319};
320
321/**
322 * Attempts to flush the packets buffer.
323 *
324 * @api private
325 */
326
327Socket.prototype.flush = function () {
328  if ('closed' != this.readyState && this.transport.writable
329    && this.writeBuffer.length) {
330    debug('flushing buffer to transport');
331    this.emit('flush', this.writeBuffer);
332    this.server.emit('flush', this, this.writeBuffer);
333    var wbuf = this.writeBuffer;
334    this.writeBuffer = [];
335    if (!this.transport.supportsFraming) {
336      this.sentCallbackFn.push(this.packetsFn);
337    } else {
338      this.sentCallbackFn.push.apply(this.sentCallbackFn, this.packetsFn);
339    }
340    this.packetsFn = [];
341    this.transport.send(wbuf);
342    this.emit('drain');
343    this.server.emit('drain', this);
344  }
345};
346
347/**
348 * Get available upgrades for this socket.
349 *
350 * @api private
351 */
352
353Socket.prototype.getAvailableUpgrades = function () {
354  var availableUpgrades = [];
355  var allUpgrades = this.server.upgrades(this.transport.name);
356  for (var i = 0, l = allUpgrades.length; i < l; ++i) {
357    var upg = allUpgrades[i];
358    if (this.server.transports.indexOf(upg) != -1) {
359      availableUpgrades.push(upg);
360    }
361  }
362  return availableUpgrades;
363};
364
365/**
366 * Closes the socket and underlying transport.
367 *
368 * @return {Socket} for chaining
369 * @api public
370 */
371
372Socket.prototype.close = function () {
373  if ('open' == this.readyState) {
374    this.readyState = 'closing';
375    var self = this;
376    this.transport.close(function () {
377      self.onClose('forced close');
378    });
379  }
380};