Repositorio del curso CCOM4030 el semestre B91 del proyecto Artesanías con el Instituto de Cultura

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