node/node_modules/socket.io/lib/transport.js

   1
   2 /*!
   3  * socket.io-node
   4  * Copyright(c) 2011 LearnBoost <dev@learnboost.com>
   5  * MIT Licensed
   6  */
   7
   8 /**
   9  * Module dependencies.
  10  */
  11
  12 var parser = require('./parser');
  13
  14 /**
  15  * Expose the constructor.
  16  */
  17
  18 exports = module.exports = Transport;
  19
  20 /**
  21  * Transport constructor.
  22  *
  23  * @api public
  24  */
  25
  26 function Transport (mng, data, req) {
  27   this.manager = mng;
  28   this.id = data.id;
  29   this.disconnected = false;
  30   this.drained = true;
  31   this.handleRequest(req);
  32 };
  33
  34 /**
  35  * Access the logger.
  36  *
  37  * @api public
  38  */
  39
  40 Transport.prototype.__defineGetter__('log', function () {
  41   return this.manager.log;
  42 });
  43
  44 /**
  45  * Access the store.
  46  *
  47  * @api public
  48  */
  49
  50 Transport.prototype.__defineGetter__('store', function () {
  51   return this.manager.store;
  52 });
  53
  54 /**
  55  * Handles a request when it's set.
  56  *
  57  * @api private
  58  */
  59
  60 Transport.prototype.handleRequest = function (req) {
  61   this.log.debug('setting request', req.method, req.url);
  62   this.req = req;
  63
  64   if (req.method == 'GET') {
  65     this.socket = req.socket;
  66     this.open = true;
  67     this.drained = true;
  68     this.setHeartbeatInterval();
  69
  70     this.setHandlers();
  71     this.onSocketConnect();
  72   }
  73 };
  74
  75 /**
  76  * Called when a connection is first set.
  77  *
  78  * @api private
  79  */
  80
  81 Transport.prototype.onSocketConnect = function () { };
  82
  83 /**
  84  * Sets transport handlers
  85  *
  86  * @api private
  87  */
  88
  89 Transport.prototype.setHandlers = function () {
  90   var self = this;
  91
  92   // we need to do this in a pub/sub way since the client can POST the message
  93   // over a different socket (ie: different Transport instance)
  94   this.store.subscribe('heartbeat-clear:' + this.id, function () {
  95     self.onHeartbeatClear();
  96   });
  97
  98   this.store.subscribe('disconnect-force:' + this.id, function () {
  99     self.onForcedDisconnect();
 100   });
 101
 102   this.store.subscribe('dispatch:' + this.id, function (packet, volatile) {
 103     self.onDispatch(packet, volatile);
 104   });
 105
 106   this.bound = {
 107       end: this.onSocketEnd.bind(this)
 108     , close: this.onSocketClose.bind(this)
 109     , error: this.onSocketError.bind(this)
 110     , drain: this.onSocketDrain.bind(this)
 111   };
 112
 113   this.socket.on('end', this.bound.end);
 114   this.socket.on('close', this.bound.close);
 115   this.socket.on('error', this.bound.error);
 116   this.socket.on('drain', this.bound.drain);
 117
 118   this.handlersSet = true;
 119 };
 120
 121 /**
 122  * Removes transport handlers
 123  *
 124  * @api private
 125  */
 126
 127 Transport.prototype.clearHandlers = function () {
 128   if (this.handlersSet) {
 129     this.store.unsubscribe('disconnect-force:' + this.id);
 130     this.store.unsubscribe('heartbeat-clear:' + this.id);
 131     this.store.unsubscribe('dispatch:' + this.id);
 132
 133     this.socket.removeListener('end', this.bound.end);
 134     this.socket.removeListener('close', this.bound.close);
 135     this.socket.removeListener('error', this.bound.error);
 136     this.socket.removeListener('drain', this.bound.drain);
 137   }
 138 };
 139
 140 /**
 141  * Called when the connection dies
 142  *
 143  * @api private
 144  */
 145
 146 Transport.prototype.onSocketEnd = function () {
 147   this.end('socket end');
 148 };
 149
 150 /**
 151  * Called when the connection dies
 152  *
 153  * @api private
 154  */
 155
 156 Transport.prototype.onSocketClose = function (error) {
 157   this.end(error ? 'socket error' : 'socket close');
 158 };
 159
 160 /**
 161  * Called when the connection has an error.
 162  *
 163  * @api private
 164  */
 165
 166 Transport.prototype.onSocketError = function (err) {
 167   if (this.open) {
 168     this.socket.destroy();
 169     this.onClose();
 170   }
 171
 172   this.log.info('socket error '  + err.stack);
 173 };
 174
 175 /**
 176  * Called when the connection is drained.
 177  *
 178  * @api private
 179  */
 180
 181 Transport.prototype.onSocketDrain = function () {
 182   this.drained = true;
 183 };
 184
 185 /**
 186  * Called upon receiving a heartbeat packet.
 187  *
 188  * @api private
 189  */
 190
 191 Transport.prototype.onHeartbeatClear = function () {
 192   this.clearHeartbeatTimeout();
 193   this.setHeartbeatInterval();
 194 };
 195
 196 /**
 197  * Called upon a forced disconnection.
 198  *
 199  * @api private
 200  */
 201
 202 Transport.prototype.onForcedDisconnect = function () {
 203   if (!this.disconnected) {
 204     this.log.info('transport end by forced client disconnection');
 205     if (this.open) {
 206       this.packet({ type: 'disconnect' });
 207     }
 208     this.end('booted');
 209   }
 210 };
 211
 212 /**
 213  * Dispatches a packet.
 214  *
 215  * @api private
 216  */
 217
 218 Transport.prototype.onDispatch = function (packet, volatile) {
 219   if (volatile) {
 220     this.writeVolatile(packet);
 221   } else {
 222     this.write(packet);
 223   }
 224 };
 225
 226 /**
 227  * Sets the close timeout.
 228  */
 229
 230 Transport.prototype.setCloseTimeout = function () {
 231   if (!this.closeTimeout) {
 232     var self = this;
 233
 234     this.closeTimeout = setTimeout(function () {
 235       self.log.debug('fired close timeout for client', self.id);
 236       self.closeTimeout = null;
 237       self.end('close timeout');
 238     }, this.manager.get('close timeout') * 1000);
 239
 240     this.log.debug('set close timeout for client', this.id);
 241   }
 242 };
 243
 244 /**
 245  * Clears the close timeout.
 246  */
 247
 248 Transport.prototype.clearCloseTimeout = function () {
 249   if (this.closeTimeout) {
 250     clearTimeout(this.closeTimeout);
 251     this.closeTimeout = null;
 252
 253     this.log.debug('cleared close timeout for client', this.id);
 254   }
 255 };
 256
 257 /**
 258  * Sets the heartbeat timeout
 259  */
 260
 261 Transport.prototype.setHeartbeatTimeout = function () {
 262   if (!this.heartbeatTimeout) {
 263     var self = this;
 264
 265     this.heartbeatTimeout = setTimeout(function () {
 266       self.log.debug('fired heartbeat timeout for client', self.id);
 267       self.heartbeatTimeout = null;
 268       self.end('heartbeat timeout');
 269     }, this.manager.get('heartbeat timeout') * 1000);
 270
 271     this.log.debug('set heartbeat timeout for client', this.id);
 272   }
 273 };
 274
 275 /**
 276  * Clears the heartbeat timeout
 277  *
 278  * @param text
 279  */
 280
 281 Transport.prototype.clearHeartbeatTimeout = function () {
 282   if (this.heartbeatTimeout) {
 283     clearTimeout(this.heartbeatTimeout);
 284     this.heartbeatTimeout = null;
 285     this.log.debug('cleared heartbeat timeout for client', this.id);
 286   }
 287 };
 288
 289 /**
 290  * Sets the heartbeat interval. To be called when a connection opens and when
 291  * a heartbeat is received.
 292  *
 293  * @api private
 294  */
 295
 296 Transport.prototype.setHeartbeatInterval = function () {
 297   if (!this.heartbeatInterval) {
 298     var self = this;
 299
 300     this.heartbeatInterval = setTimeout(function () {
 301       self.heartbeat();
 302       self.heartbeatInterval = null;
 303     }, this.manager.get('heartbeat interval') * 1000);
 304
 305     this.log.debug('set heartbeat interval for client', this.id);
 306   }
 307 };
 308
 309 /**
 310  * Clears all timeouts.
 311  *
 312  * @api private
 313  */
 314
 315 Transport.prototype.clearTimeouts = function () {
 316   this.clearCloseTimeout();
 317   this.clearHeartbeatTimeout();
 318   this.clearHeartbeatInterval();
 319 };
 320
 321 /**
 322  * Sends a heartbeat
 323  *
 324  * @api private
 325  */
 326
 327 Transport.prototype.heartbeat = function () {
 328   if (this.open) {
 329     this.log.debug('emitting heartbeat for client', this.id);
 330     this.packet({ type: 'heartbeat' });
 331     this.setHeartbeatTimeout();
 332   }
 333
 334   return this;
 335 };
 336
 337 /**
 338  * Handles a message.
 339  *
 340  * @param {Object} packet object
 341  * @api private
 342  */
 343
 344 Transport.prototype.onMessage = function (packet) {
 345   var current = this.manager.transports[this.id];
 346
 347   if ('heartbeat' == packet.type) {
 348     this.log.debug('got heartbeat packet');
 349
 350     if (current && current.open) {
 351       current.onHeartbeatClear();
 352     } else {
 353       this.store.publish('heartbeat-clear:' + this.id);
 354     }
 355   } else {
 356     if ('disconnect' == packet.type && packet.endpoint == '') {
 357       this.log.debug('got disconnection packet');
 358
 359       if (current) {
 360         current.onForcedDisconnect();
 361       } else {
 362         this.store.publish('disconnect-force:' + this.id);
 363       }
 364
 365       return;
 366     }
 367
 368     if (packet.id && packet.ack != 'data') {
 369       this.log.debug('acknowledging packet automatically');
 370
 371       var ack = parser.encodePacket({
 372           type: 'ack'
 373         , ackId: packet.id
 374         , endpoint: packet.endpoint || ''
 375       });
 376
 377       if (current && current.open) {
 378         current.onDispatch(ack);
 379       } else {
 380         this.manager.onClientDispatch(this.id, ack);
 381         this.store.publish('dispatch:' + this.id, ack);
 382       }
 383     }
 384
 385     // handle packet locally or publish it
 386     if (current) {
 387       this.manager.onClientMessage(this.id, packet);
 388     } else {
 389       this.store.publish('message:' + this.id, packet);
 390     }
 391   }
 392 };
 393
 394 /**
 395  * Clears the heartbeat interval
 396  *
 397  * @api private
 398  */
 399
 400 Transport.prototype.clearHeartbeatInterval = function () {
 401   if (this.heartbeatInterval) {
 402     clearTimeout(this.heartbeatInterval);
 403     this.heartbeatInterval = null;
 404     this.log.debug('cleared heartbeat interval for client', this.id);
 405   }
 406 };
 407
 408 /**
 409  * Finishes the connection and makes sure client doesn't reopen
 410  *
 411  * @api private
 412  */
 413
 414 Transport.prototype.disconnect = function (reason) {
 415   this.packet({ type: 'disconnect' });
 416   this.end(reason);
 417
 418   return this;
 419 };
 420
 421 /**
 422  * Closes the connection.
 423  *
 424  * @api private
 425  */
 426
 427 Transport.prototype.close = function () {
 428   if (this.open) {
 429     this.doClose();
 430     this.onClose();
 431   }
 432 };
 433
 434 /**
 435  * Called upon a connection close.
 436  *
 437  * @api private
 438  */
 439
 440 Transport.prototype.onClose = function () {
 441   if (this.open) {
 442     this.setCloseTimeout();
 443     this.clearHandlers();
 444     this.open = false;
 445     this.manager.onClose(this.id);
 446     this.store.publish('close', this.id);
 447   }
 448 };
 449
 450 /**
 451  * Cleans up the connection, considers the client disconnected.
 452  *
 453  * @api private
 454  */
 455
 456 Transport.prototype.end = function (reason) {
 457   if (!this.disconnected) {
 458     this.log.info('transport end');
 459
 460     var local = this.manager.transports[this.id];
 461
 462     this.close();
 463     this.clearTimeouts();
 464     this.disconnected = true;
 465
 466     if (local) {
 467       this.manager.onClientDisconnect(this.id, reason, true);
 468     } else {
 469       this.store.publish('disconnect:' + this.id, reason);
 470     }
 471   }
 472 };
 473
 474 /**
 475  * Signals that the transport should pause and buffer data.
 476  *
 477  * @api public
 478  */
 479
 480 Transport.prototype.discard = function () {
 481   this.log.debug('discarding transport');
 482   this.discarded = true;
 483   this.clearTimeouts();
 484   this.clearHandlers();
 485
 486   return this;
 487 };
 488
 489 /**
 490  * Writes an error packet with the specified reason and advice.
 491  *
 492  * @param {Number} advice
 493  * @param {Number} reason
 494  * @api public
 495  */
 496
 497 Transport.prototype.error = function (reason, advice) {
 498   this.packet({
 499       type: 'error'
 500     , reason: reason
 501     , advice: advice
 502   });
 503
 504   this.log.warn(reason, advice ? ('client should ' + advice) : '');
 505   this.end('error');
 506 };
 507
 508 /**
 509  * Write a packet.
 510  *
 511  * @api public
 512  */
 513
 514 Transport.prototype.packet = function (obj) {
 515   return this.write(parser.encodePacket(obj));
 516 };
 517
 518 /**
 519  * Writes a volatile message.
 520  *
 521  * @api private
 522  */
 523
 524 Transport.prototype.writeVolatile = function (msg) {
 525   if (this.open) {
 526     if (this.drained) {
 527       this.write(msg);
 528     } else {
 529       this.log.debug('ignoring volatile packet, buffer not drained');
 530     }
 531   } else {
 532     this.log.debug('ignoring volatile packet, transport not open');
 533   }
 534 };