socket.js 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460
  1. "use strict";
  2. Object.defineProperty(exports, "__esModule", { value: true });
  3. exports.Socket = void 0;
  4. const events_1 = require("events");
  5. const debug_1 = require("debug");
  6. const timers_1 = require("timers");
  7. const debug = (0, debug_1.default)("engine:socket");
  8. class Socket extends events_1.EventEmitter {
  9. get readyState() {
  10. return this._readyState;
  11. }
  12. set readyState(state) {
  13. debug("readyState updated from %s to %s", this._readyState, state);
  14. this._readyState = state;
  15. }
  16. constructor(id, server, transport, req, protocol) {
  17. super();
  18. /**
  19. * The current state of the socket.
  20. */
  21. this._readyState = "opening";
  22. /* private */ this.upgrading = false;
  23. /* private */ this.upgraded = false;
  24. this.writeBuffer = [];
  25. this.packetsFn = [];
  26. this.sentCallbackFn = [];
  27. this.cleanupFn = [];
  28. this.id = id;
  29. this.server = server;
  30. this.request = req;
  31. this.protocol = protocol;
  32. // Cache IP since it might not be in the req later
  33. if (req) {
  34. if (req.websocket && req.websocket._socket) {
  35. this.remoteAddress = req.websocket._socket.remoteAddress;
  36. }
  37. else {
  38. this.remoteAddress = req.connection.remoteAddress;
  39. }
  40. }
  41. else {
  42. // TODO there is currently no way to get the IP address of the client when it connects with WebTransport
  43. // see https://github.com/fails-components/webtransport/issues/114
  44. }
  45. this.pingTimeoutTimer = null;
  46. this.pingIntervalTimer = null;
  47. this.setTransport(transport);
  48. this.onOpen();
  49. }
  50. /**
  51. * Called upon transport considered open.
  52. *
  53. * @private
  54. */
  55. onOpen() {
  56. this.readyState = "open";
  57. // sends an `open` packet
  58. this.transport.sid = this.id;
  59. this.sendPacket("open", JSON.stringify({
  60. sid: this.id,
  61. upgrades: this.getAvailableUpgrades(),
  62. pingInterval: this.server.opts.pingInterval,
  63. pingTimeout: this.server.opts.pingTimeout,
  64. maxPayload: this.server.opts.maxHttpBufferSize,
  65. }));
  66. if (this.server.opts.initialPacket) {
  67. this.sendPacket("message", this.server.opts.initialPacket);
  68. }
  69. this.emit("open");
  70. if (this.protocol === 3) {
  71. // in protocol v3, the client sends a ping, and the server answers with a pong
  72. this.resetPingTimeout();
  73. }
  74. else {
  75. // in protocol v4, the server sends a ping, and the client answers with a pong
  76. this.schedulePing();
  77. }
  78. }
  79. /**
  80. * Called upon transport packet.
  81. *
  82. * @param {Object} packet
  83. * @private
  84. */
  85. onPacket(packet) {
  86. if ("open" !== this.readyState) {
  87. return debug("packet received with closed socket");
  88. }
  89. // export packet event
  90. debug(`received packet ${packet.type}`);
  91. this.emit("packet", packet);
  92. switch (packet.type) {
  93. case "ping":
  94. if (this.transport.protocol !== 3) {
  95. this.onError(new Error("invalid heartbeat direction"));
  96. return;
  97. }
  98. debug("got ping");
  99. this.pingTimeoutTimer.refresh();
  100. this.sendPacket("pong");
  101. this.emit("heartbeat");
  102. break;
  103. case "pong":
  104. if (this.transport.protocol === 3) {
  105. this.onError(new Error("invalid heartbeat direction"));
  106. return;
  107. }
  108. debug("got pong");
  109. (0, timers_1.clearTimeout)(this.pingTimeoutTimer);
  110. this.pingIntervalTimer.refresh();
  111. this.emit("heartbeat");
  112. break;
  113. case "error":
  114. this.onClose("parse error");
  115. break;
  116. case "message":
  117. this.emit("data", packet.data);
  118. this.emit("message", packet.data);
  119. break;
  120. }
  121. }
  122. /**
  123. * Called upon transport error.
  124. *
  125. * @param {Error} err - error object
  126. * @private
  127. */
  128. onError(err) {
  129. debug("transport error");
  130. this.onClose("transport error", err);
  131. }
  132. /**
  133. * Pings client every `this.pingInterval` and expects response
  134. * within `this.pingTimeout` or closes connection.
  135. *
  136. * @private
  137. */
  138. schedulePing() {
  139. this.pingIntervalTimer = (0, timers_1.setTimeout)(() => {
  140. debug("writing ping packet - expecting pong within %sms", this.server.opts.pingTimeout);
  141. this.sendPacket("ping");
  142. this.resetPingTimeout();
  143. }, this.server.opts.pingInterval);
  144. }
  145. /**
  146. * Resets ping timeout.
  147. *
  148. * @private
  149. */
  150. resetPingTimeout() {
  151. (0, timers_1.clearTimeout)(this.pingTimeoutTimer);
  152. this.pingTimeoutTimer = (0, timers_1.setTimeout)(() => {
  153. if (this.readyState === "closed")
  154. return;
  155. this.onClose("ping timeout");
  156. }, this.protocol === 3
  157. ? this.server.opts.pingInterval + this.server.opts.pingTimeout
  158. : this.server.opts.pingTimeout);
  159. }
  160. /**
  161. * Attaches handlers for the given transport.
  162. *
  163. * @param {Transport} transport
  164. * @private
  165. */
  166. setTransport(transport) {
  167. const onError = this.onError.bind(this);
  168. const onReady = () => this.flush();
  169. const onPacket = this.onPacket.bind(this);
  170. const onDrain = this.onDrain.bind(this);
  171. const onClose = this.onClose.bind(this, "transport close");
  172. this.transport = transport;
  173. this.transport.once("error", onError);
  174. this.transport.on("ready", onReady);
  175. this.transport.on("packet", onPacket);
  176. this.transport.on("drain", onDrain);
  177. this.transport.once("close", onClose);
  178. this.cleanupFn.push(function () {
  179. transport.removeListener("error", onError);
  180. transport.removeListener("ready", onReady);
  181. transport.removeListener("packet", onPacket);
  182. transport.removeListener("drain", onDrain);
  183. transport.removeListener("close", onClose);
  184. });
  185. }
  186. /**
  187. * Upon transport "drain" event
  188. *
  189. * @private
  190. */
  191. onDrain() {
  192. if (this.sentCallbackFn.length > 0) {
  193. debug("executing batch send callback");
  194. const seqFn = this.sentCallbackFn.shift();
  195. if (seqFn) {
  196. for (let i = 0; i < seqFn.length; i++) {
  197. seqFn[i](this.transport);
  198. }
  199. }
  200. }
  201. }
  202. /**
  203. * Upgrades socket to the given transport
  204. *
  205. * @param {Transport} transport
  206. * @private
  207. */
  208. /* private */ _maybeUpgrade(transport) {
  209. debug('might upgrade socket transport from "%s" to "%s"', this.transport.name, transport.name);
  210. this.upgrading = true;
  211. // set transport upgrade timer
  212. const upgradeTimeoutTimer = (0, timers_1.setTimeout)(() => {
  213. debug("client did not complete upgrade - closing transport");
  214. cleanup();
  215. if ("open" === transport.readyState) {
  216. transport.close();
  217. }
  218. }, this.server.opts.upgradeTimeout);
  219. let checkIntervalTimer;
  220. const onPacket = (packet) => {
  221. if ("ping" === packet.type && "probe" === packet.data) {
  222. debug("got probe ping packet, sending pong");
  223. transport.send([{ type: "pong", data: "probe" }]);
  224. this.emit("upgrading", transport);
  225. clearInterval(checkIntervalTimer);
  226. checkIntervalTimer = setInterval(check, 100);
  227. }
  228. else if ("upgrade" === packet.type && this.readyState !== "closed") {
  229. debug("got upgrade packet - upgrading");
  230. cleanup();
  231. this.transport.discard();
  232. this.upgraded = true;
  233. this.clearTransport();
  234. this.setTransport(transport);
  235. this.emit("upgrade", transport);
  236. this.flush();
  237. if (this.readyState === "closing") {
  238. transport.close(() => {
  239. this.onClose("forced close");
  240. });
  241. }
  242. }
  243. else {
  244. cleanup();
  245. transport.close();
  246. }
  247. };
  248. // we force a polling cycle to ensure a fast upgrade
  249. const check = () => {
  250. if ("polling" === this.transport.name && this.transport.writable) {
  251. debug("writing a noop packet to polling for fast upgrade");
  252. this.transport.send([{ type: "noop" }]);
  253. }
  254. };
  255. const cleanup = () => {
  256. this.upgrading = false;
  257. clearInterval(checkIntervalTimer);
  258. (0, timers_1.clearTimeout)(upgradeTimeoutTimer);
  259. transport.removeListener("packet", onPacket);
  260. transport.removeListener("close", onTransportClose);
  261. transport.removeListener("error", onError);
  262. this.removeListener("close", onClose);
  263. };
  264. const onError = (err) => {
  265. debug("client did not complete upgrade - %s", err);
  266. cleanup();
  267. transport.close();
  268. transport = null;
  269. };
  270. const onTransportClose = () => {
  271. onError("transport closed");
  272. };
  273. const onClose = () => {
  274. onError("socket closed");
  275. };
  276. transport.on("packet", onPacket);
  277. transport.once("close", onTransportClose);
  278. transport.once("error", onError);
  279. this.once("close", onClose);
  280. }
  281. /**
  282. * Clears listeners and timers associated with current transport.
  283. *
  284. * @private
  285. */
  286. clearTransport() {
  287. let cleanup;
  288. const toCleanUp = this.cleanupFn.length;
  289. for (let i = 0; i < toCleanUp; i++) {
  290. cleanup = this.cleanupFn.shift();
  291. cleanup();
  292. }
  293. // silence further transport errors and prevent uncaught exceptions
  294. this.transport.on("error", function () {
  295. debug("error triggered by discarded transport");
  296. });
  297. // ensure transport won't stay open
  298. this.transport.close();
  299. (0, timers_1.clearTimeout)(this.pingTimeoutTimer);
  300. }
  301. /**
  302. * Called upon transport considered closed.
  303. * Possible reasons: `ping timeout`, `client error`, `parse error`,
  304. * `transport error`, `server close`, `transport close`
  305. */
  306. onClose(reason, description) {
  307. if ("closed" !== this.readyState) {
  308. this.readyState = "closed";
  309. // clear timers
  310. (0, timers_1.clearTimeout)(this.pingIntervalTimer);
  311. (0, timers_1.clearTimeout)(this.pingTimeoutTimer);
  312. // clean writeBuffer in next tick, so developers can still
  313. // grab the writeBuffer on 'close' event
  314. process.nextTick(() => {
  315. this.writeBuffer = [];
  316. });
  317. this.packetsFn = [];
  318. this.sentCallbackFn = [];
  319. this.clearTransport();
  320. this.emit("close", reason, description);
  321. }
  322. }
  323. /**
  324. * Sends a message packet.
  325. *
  326. * @param {Object} data
  327. * @param {Object} options
  328. * @param {Function} callback
  329. * @return {Socket} for chaining
  330. */
  331. send(data, options, callback) {
  332. this.sendPacket("message", data, options, callback);
  333. return this;
  334. }
  335. /**
  336. * Alias of {@link send}.
  337. *
  338. * @param data
  339. * @param options
  340. * @param callback
  341. */
  342. write(data, options, callback) {
  343. this.sendPacket("message", data, options, callback);
  344. return this;
  345. }
  346. /**
  347. * Sends a packet.
  348. *
  349. * @param {String} type - packet type
  350. * @param {String} data
  351. * @param {Object} options
  352. * @param {Function} callback
  353. *
  354. * @private
  355. */
  356. sendPacket(type, data, options = {}, callback) {
  357. if ("function" === typeof options) {
  358. callback = options;
  359. options = {};
  360. }
  361. if ("closing" !== this.readyState && "closed" !== this.readyState) {
  362. debug('sending packet "%s" (%s)', type, data);
  363. // compression is enabled by default
  364. options.compress = options.compress !== false;
  365. const packet = {
  366. type,
  367. options: options,
  368. };
  369. if (data)
  370. packet.data = data;
  371. // exports packetCreate event
  372. this.emit("packetCreate", packet);
  373. this.writeBuffer.push(packet);
  374. // add send callback to object, if defined
  375. if ("function" === typeof callback)
  376. this.packetsFn.push(callback);
  377. this.flush();
  378. }
  379. }
  380. /**
  381. * Attempts to flush the packets buffer.
  382. *
  383. * @private
  384. */
  385. flush() {
  386. if ("closed" !== this.readyState &&
  387. this.transport.writable &&
  388. this.writeBuffer.length) {
  389. debug("flushing buffer to transport");
  390. this.emit("flush", this.writeBuffer);
  391. this.server.emit("flush", this, this.writeBuffer);
  392. const wbuf = this.writeBuffer;
  393. this.writeBuffer = [];
  394. if (this.packetsFn.length) {
  395. this.sentCallbackFn.push(this.packetsFn);
  396. this.packetsFn = [];
  397. }
  398. else {
  399. this.sentCallbackFn.push(null);
  400. }
  401. this.transport.send(wbuf);
  402. this.emit("drain");
  403. this.server.emit("drain", this);
  404. }
  405. }
  406. /**
  407. * Get available upgrades for this socket.
  408. *
  409. * @private
  410. */
  411. getAvailableUpgrades() {
  412. const availableUpgrades = [];
  413. const allUpgrades = this.server.upgrades(this.transport.name);
  414. for (let i = 0; i < allUpgrades.length; ++i) {
  415. const upg = allUpgrades[i];
  416. if (this.server.opts.transports.indexOf(upg) !== -1) {
  417. availableUpgrades.push(upg);
  418. }
  419. }
  420. return availableUpgrades;
  421. }
  422. /**
  423. * Closes the socket and underlying transport.
  424. *
  425. * @param {Boolean} discard - optional, discard the transport
  426. * @return {Socket} for chaining
  427. */
  428. close(discard) {
  429. if (discard &&
  430. (this.readyState === "open" || this.readyState === "closing")) {
  431. return this.closeTransport(discard);
  432. }
  433. if ("open" !== this.readyState)
  434. return;
  435. this.readyState = "closing";
  436. if (this.writeBuffer.length) {
  437. debug("there are %d remaining packets in the buffer, waiting for the 'drain' event", this.writeBuffer.length);
  438. this.once("drain", () => {
  439. debug("all packets have been sent, closing the transport");
  440. this.closeTransport(discard);
  441. });
  442. return;
  443. }
  444. debug("the buffer is empty, closing the transport right away");
  445. this.closeTransport(discard);
  446. }
  447. /**
  448. * Closes the underlying transport.
  449. *
  450. * @param {Boolean} discard
  451. * @private
  452. */
  453. closeTransport(discard) {
  454. debug("closing the transport (discard? %s)", !!discard);
  455. if (discard)
  456. this.transport.discard();
  457. this.transport.close(this.onClose.bind(this, "forced close"));
  458. }
  459. }
  460. exports.Socket = Socket;