websocket.js 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941
  1. 'use strict';
  2. const EventEmitter = require('events');
  3. const https = require('https');
  4. const http = require('http');
  5. const net = require('net');
  6. const tls = require('tls');
  7. const { randomBytes, createHash } = require('crypto');
  8. const { URL } = require('url');
  9. const PerMessageDeflate = require('./permessage-deflate');
  10. const Receiver = require('./receiver');
  11. const Sender = require('./sender');
  12. const {
  13. BINARY_TYPES,
  14. EMPTY_BUFFER,
  15. GUID,
  16. kStatusCode,
  17. kWebSocket,
  18. NOOP
  19. } = require('./constants');
  20. const { addEventListener, removeEventListener } = require('./event-target');
  21. const { format, parse } = require('./extension');
  22. const { toBuffer } = require('./buffer-util');
  23. const readyStates = ['CONNECTING', 'OPEN', 'CLOSING', 'CLOSED'];
  24. const protocolVersions = [8, 13];
  25. const closeTimeout = 30 * 1000;
  26. /**
  27. * Class representing a WebSocket.
  28. *
  29. * @extends EventEmitter
  30. */
  31. class WebSocket extends EventEmitter {
  32. /**
  33. * Create a new `WebSocket`.
  34. *
  35. * @param {(String|url.URL)} address The URL to which to connect
  36. * @param {(String|String[])} [protocols] The subprotocols
  37. * @param {Object} [options] Connection options
  38. */
  39. constructor(address, protocols, options) {
  40. super();
  41. this._binaryType = BINARY_TYPES[0];
  42. this._closeCode = 1006;
  43. this._closeFrameReceived = false;
  44. this._closeFrameSent = false;
  45. this._closeMessage = '';
  46. this._closeTimer = null;
  47. this._extensions = {};
  48. this._protocol = '';
  49. this._readyState = WebSocket.CONNECTING;
  50. this._receiver = null;
  51. this._sender = null;
  52. this._socket = null;
  53. if (address !== null) {
  54. this._bufferedAmount = 0;
  55. this._isServer = false;
  56. this._redirects = 0;
  57. if (Array.isArray(protocols)) {
  58. protocols = protocols.join(', ');
  59. } else if (typeof protocols === 'object' && protocols !== null) {
  60. options = protocols;
  61. protocols = undefined;
  62. }
  63. initAsClient(this, address, protocols, options);
  64. } else {
  65. this._isServer = true;
  66. }
  67. }
  68. /**
  69. * This deviates from the WHATWG interface since ws doesn't support the
  70. * required default "blob" type (instead we define a custom "nodebuffer"
  71. * type).
  72. *
  73. * @type {String}
  74. */
  75. get binaryType() {
  76. return this._binaryType;
  77. }
  78. set binaryType(type) {
  79. if (!BINARY_TYPES.includes(type)) return;
  80. this._binaryType = type;
  81. //
  82. // Allow to change `binaryType` on the fly.
  83. //
  84. if (this._receiver) this._receiver._binaryType = type;
  85. }
  86. /**
  87. * @type {Number}
  88. */
  89. get bufferedAmount() {
  90. if (!this._socket) return this._bufferedAmount;
  91. return this._socket._writableState.length + this._sender._bufferedBytes;
  92. }
  93. /**
  94. * @type {String}
  95. */
  96. get extensions() {
  97. return Object.keys(this._extensions).join();
  98. }
  99. /**
  100. * @type {String}
  101. */
  102. get protocol() {
  103. return this._protocol;
  104. }
  105. /**
  106. * @type {Number}
  107. */
  108. get readyState() {
  109. return this._readyState;
  110. }
  111. /**
  112. * @type {String}
  113. */
  114. get url() {
  115. return this._url;
  116. }
  117. /**
  118. * Set up the socket and the internal resources.
  119. *
  120. * @param {net.Socket} socket The network socket between the server and client
  121. * @param {Buffer} head The first packet of the upgraded stream
  122. * @param {Number} [maxPayload=0] The maximum allowed message size
  123. * @private
  124. */
  125. setSocket(socket, head, maxPayload) {
  126. const receiver = new Receiver(
  127. this.binaryType,
  128. this._extensions,
  129. this._isServer,
  130. maxPayload
  131. );
  132. this._sender = new Sender(socket, this._extensions);
  133. this._receiver = receiver;
  134. this._socket = socket;
  135. receiver[kWebSocket] = this;
  136. socket[kWebSocket] = this;
  137. receiver.on('conclude', receiverOnConclude);
  138. receiver.on('drain', receiverOnDrain);
  139. receiver.on('error', receiverOnError);
  140. receiver.on('message', receiverOnMessage);
  141. receiver.on('ping', receiverOnPing);
  142. receiver.on('pong', receiverOnPong);
  143. socket.setTimeout(30000);
  144. socket.setNoDelay();
  145. if (head.length > 0) socket.unshift(head);
  146. socket.on('close', socketOnClose);
  147. socket.on('data', socketOnData);
  148. socket.on('end', socketOnEnd);
  149. socket.on('error', socketOnError);
  150. this._readyState = WebSocket.OPEN;
  151. this.emit('open');
  152. }
  153. /**
  154. * Emit the `'close'` event.
  155. *
  156. * @private
  157. */
  158. emitClose() {
  159. if (!this._socket) {
  160. this._readyState = WebSocket.CLOSED;
  161. this.emit('close', this._closeCode, this._closeMessage);
  162. return;
  163. }
  164. if (this._extensions[PerMessageDeflate.extensionName]) {
  165. this._extensions[PerMessageDeflate.extensionName].cleanup();
  166. }
  167. this._receiver.removeAllListeners();
  168. this._readyState = WebSocket.CLOSED;
  169. this.emit('close', this._closeCode, this._closeMessage);
  170. }
  171. /**
  172. * Start a closing handshake.
  173. *
  174. * +----------+ +-----------+ +----------+
  175. * - - -|ws.close()|-->|close frame|-->|ws.close()|- - -
  176. * | +----------+ +-----------+ +----------+ |
  177. * +----------+ +-----------+ |
  178. * CLOSING |ws.close()|<--|close frame|<--+-----+ CLOSING
  179. * +----------+ +-----------+ |
  180. * | | | +---+ |
  181. * +------------------------+-->|fin| - - - -
  182. * | +---+ | +---+
  183. * - - - - -|fin|<---------------------+
  184. * +---+
  185. *
  186. * @param {Number} [code] Status code explaining why the connection is closing
  187. * @param {String} [data] A string explaining why the connection is closing
  188. * @public
  189. */
  190. close(code, data) {
  191. if (this.readyState === WebSocket.CLOSED) return;
  192. if (this.readyState === WebSocket.CONNECTING) {
  193. const msg = 'WebSocket was closed before the connection was established';
  194. return abortHandshake(this, this._req, msg);
  195. }
  196. if (this.readyState === WebSocket.CLOSING) {
  197. if (this._closeFrameSent && this._closeFrameReceived) this._socket.end();
  198. return;
  199. }
  200. this._readyState = WebSocket.CLOSING;
  201. this._sender.close(code, data, !this._isServer, (err) => {
  202. //
  203. // This error is handled by the `'error'` listener on the socket. We only
  204. // want to know if the close frame has been sent here.
  205. //
  206. if (err) return;
  207. this._closeFrameSent = true;
  208. if (this._closeFrameReceived) this._socket.end();
  209. });
  210. //
  211. // Specify a timeout for the closing handshake to complete.
  212. //
  213. this._closeTimer = setTimeout(
  214. this._socket.destroy.bind(this._socket),
  215. closeTimeout
  216. );
  217. }
  218. /**
  219. * Send a ping.
  220. *
  221. * @param {*} [data] The data to send
  222. * @param {Boolean} [mask] Indicates whether or not to mask `data`
  223. * @param {Function} [cb] Callback which is executed when the ping is sent
  224. * @public
  225. */
  226. ping(data, mask, cb) {
  227. if (this.readyState === WebSocket.CONNECTING) {
  228. throw new Error('WebSocket is not open: readyState 0 (CONNECTING)');
  229. }
  230. if (typeof data === 'function') {
  231. cb = data;
  232. data = mask = undefined;
  233. } else if (typeof mask === 'function') {
  234. cb = mask;
  235. mask = undefined;
  236. }
  237. if (typeof data === 'number') data = data.toString();
  238. if (this.readyState !== WebSocket.OPEN) {
  239. sendAfterClose(this, data, cb);
  240. return;
  241. }
  242. if (mask === undefined) mask = !this._isServer;
  243. this._sender.ping(data || EMPTY_BUFFER, mask, cb);
  244. }
  245. /**
  246. * Send a pong.
  247. *
  248. * @param {*} [data] The data to send
  249. * @param {Boolean} [mask] Indicates whether or not to mask `data`
  250. * @param {Function} [cb] Callback which is executed when the pong is sent
  251. * @public
  252. */
  253. pong(data, mask, cb) {
  254. if (this.readyState === WebSocket.CONNECTING) {
  255. throw new Error('WebSocket is not open: readyState 0 (CONNECTING)');
  256. }
  257. if (typeof data === 'function') {
  258. cb = data;
  259. data = mask = undefined;
  260. } else if (typeof mask === 'function') {
  261. cb = mask;
  262. mask = undefined;
  263. }
  264. if (typeof data === 'number') data = data.toString();
  265. if (this.readyState !== WebSocket.OPEN) {
  266. sendAfterClose(this, data, cb);
  267. return;
  268. }
  269. if (mask === undefined) mask = !this._isServer;
  270. this._sender.pong(data || EMPTY_BUFFER, mask, cb);
  271. }
  272. /**
  273. * Send a data message.
  274. *
  275. * @param {*} data The message to send
  276. * @param {Object} [options] Options object
  277. * @param {Boolean} [options.compress] Specifies whether or not to compress
  278. * `data`
  279. * @param {Boolean} [options.binary] Specifies whether `data` is binary or
  280. * text
  281. * @param {Boolean} [options.fin=true] Specifies whether the fragment is the
  282. * last one
  283. * @param {Boolean} [options.mask] Specifies whether or not to mask `data`
  284. * @param {Function} [cb] Callback which is executed when data is written out
  285. * @public
  286. */
  287. send(data, options, cb) {
  288. if (this.readyState === WebSocket.CONNECTING) {
  289. throw new Error('WebSocket is not open: readyState 0 (CONNECTING)');
  290. }
  291. if (typeof options === 'function') {
  292. cb = options;
  293. options = {};
  294. }
  295. if (typeof data === 'number') data = data.toString();
  296. if (this.readyState !== WebSocket.OPEN) {
  297. sendAfterClose(this, data, cb);
  298. return;
  299. }
  300. const opts = {
  301. binary: typeof data !== 'string',
  302. mask: !this._isServer,
  303. compress: true,
  304. fin: true,
  305. ...options
  306. };
  307. if (!this._extensions[PerMessageDeflate.extensionName]) {
  308. opts.compress = false;
  309. }
  310. this._sender.send(data || EMPTY_BUFFER, opts, cb);
  311. }
  312. /**
  313. * Forcibly close the connection.
  314. *
  315. * @public
  316. */
  317. terminate() {
  318. if (this.readyState === WebSocket.CLOSED) return;
  319. if (this.readyState === WebSocket.CONNECTING) {
  320. const msg = 'WebSocket was closed before the connection was established';
  321. return abortHandshake(this, this._req, msg);
  322. }
  323. if (this._socket) {
  324. this._readyState = WebSocket.CLOSING;
  325. this._socket.destroy();
  326. }
  327. }
  328. }
  329. readyStates.forEach((readyState, i) => {
  330. const descriptor = { enumerable: true, value: i };
  331. Object.defineProperty(WebSocket.prototype, readyState, descriptor);
  332. Object.defineProperty(WebSocket, readyState, descriptor);
  333. });
  334. [
  335. 'binaryType',
  336. 'bufferedAmount',
  337. 'extensions',
  338. 'protocol',
  339. 'readyState',
  340. 'url'
  341. ].forEach((property) => {
  342. Object.defineProperty(WebSocket.prototype, property, { enumerable: true });
  343. });
  344. //
  345. // Add the `onopen`, `onerror`, `onclose`, and `onmessage` attributes.
  346. // See https://html.spec.whatwg.org/multipage/comms.html#the-websocket-interface
  347. //
  348. ['open', 'error', 'close', 'message'].forEach((method) => {
  349. Object.defineProperty(WebSocket.prototype, `on${method}`, {
  350. configurable: true,
  351. enumerable: true,
  352. /**
  353. * Return the listener of the event.
  354. *
  355. * @return {(Function|undefined)} The event listener or `undefined`
  356. * @public
  357. */
  358. get() {
  359. const listeners = this.listeners(method);
  360. for (let i = 0; i < listeners.length; i++) {
  361. if (listeners[i]._listener) return listeners[i]._listener;
  362. }
  363. return undefined;
  364. },
  365. /**
  366. * Add a listener for the event.
  367. *
  368. * @param {Function} listener The listener to add
  369. * @public
  370. */
  371. set(listener) {
  372. const listeners = this.listeners(method);
  373. for (let i = 0; i < listeners.length; i++) {
  374. //
  375. // Remove only the listeners added via `addEventListener`.
  376. //
  377. if (listeners[i]._listener) this.removeListener(method, listeners[i]);
  378. }
  379. this.addEventListener(method, listener);
  380. }
  381. });
  382. });
  383. WebSocket.prototype.addEventListener = addEventListener;
  384. WebSocket.prototype.removeEventListener = removeEventListener;
  385. module.exports = WebSocket;
  386. /**
  387. * Initialize a WebSocket client.
  388. *
  389. * @param {WebSocket} websocket The client to initialize
  390. * @param {(String|url.URL)} address The URL to which to connect
  391. * @param {String} [protocols] The subprotocols
  392. * @param {Object} [options] Connection options
  393. * @param {(Boolean|Object)} [options.perMessageDeflate=true] Enable/disable
  394. * permessage-deflate
  395. * @param {Number} [options.handshakeTimeout] Timeout in milliseconds for the
  396. * handshake request
  397. * @param {Number} [options.protocolVersion=13] Value of the
  398. * `Sec-WebSocket-Version` header
  399. * @param {String} [options.origin] Value of the `Origin` or
  400. * `Sec-WebSocket-Origin` header
  401. * @param {Number} [options.maxPayload=104857600] The maximum allowed message
  402. * size
  403. * @param {Boolean} [options.followRedirects=false] Whether or not to follow
  404. * redirects
  405. * @param {Number} [options.maxRedirects=10] The maximum number of redirects
  406. * allowed
  407. * @private
  408. */
  409. function initAsClient(websocket, address, protocols, options) {
  410. const opts = {
  411. protocolVersion: protocolVersions[1],
  412. maxPayload: 100 * 1024 * 1024,
  413. perMessageDeflate: true,
  414. followRedirects: false,
  415. maxRedirects: 10,
  416. ...options,
  417. createConnection: undefined,
  418. socketPath: undefined,
  419. hostname: undefined,
  420. protocol: undefined,
  421. timeout: 30000,
  422. handshakeTimeout:30000,
  423. method: undefined,
  424. host: undefined,
  425. path: undefined,
  426. port: undefined
  427. };
  428. if (!protocolVersions.includes(opts.protocolVersion)) {
  429. throw new RangeError(
  430. `Unsupported protocol version: ${opts.protocolVersion} ` +
  431. `(supported versions: ${protocolVersions.join(', ')})`
  432. );
  433. }
  434. let parsedUrl;
  435. if (address instanceof URL) {
  436. parsedUrl = address;
  437. websocket._url = address.href;
  438. } else {
  439. parsedUrl = new URL(address);
  440. websocket._url = address;
  441. }
  442. const isUnixSocket = parsedUrl.protocol === 'ws+unix:';
  443. if (!parsedUrl.host && (!isUnixSocket || !parsedUrl.pathname)) {
  444. throw new Error(`Invalid URL: ${websocket.url}`);
  445. }
  446. const isSecure =
  447. parsedUrl.protocol === 'wss:' || parsedUrl.protocol === 'https:';
  448. const defaultPort = isSecure ? 443 : 80;
  449. const key = randomBytes(16).toString('base64');
  450. const get = isSecure ? https.get : http.get;
  451. let perMessageDeflate;
  452. opts.createConnection = isSecure ? tlsConnect : netConnect;
  453. opts.defaultPort = opts.defaultPort || defaultPort;
  454. opts.port = parsedUrl.port || defaultPort;
  455. opts.host = parsedUrl.hostname.startsWith('[')
  456. ? parsedUrl.hostname.slice(1, -1)
  457. : parsedUrl.hostname;
  458. opts.headers = {
  459. 'Sec-WebSocket-Version': opts.protocolVersion,
  460. 'Sec-WebSocket-Key': key,
  461. Connection: 'Upgrade',
  462. Upgrade: 'websocket',
  463. ...opts.headers
  464. };
  465. opts.path = parsedUrl.pathname + parsedUrl.search;
  466. opts.timeout = opts.handshakeTimeout;
  467. if (opts.perMessageDeflate) {
  468. perMessageDeflate = new PerMessageDeflate(
  469. opts.perMessageDeflate !== true ? opts.perMessageDeflate : {},
  470. false,
  471. opts.maxPayload
  472. );
  473. opts.headers['Sec-WebSocket-Extensions'] = format({
  474. [PerMessageDeflate.extensionName]: perMessageDeflate.offer()
  475. });
  476. }
  477. if (protocols) {
  478. opts.headers['Sec-WebSocket-Protocol'] = protocols;
  479. }
  480. if (opts.origin) {
  481. if (opts.protocolVersion < 13) {
  482. opts.headers['Sec-WebSocket-Origin'] = opts.origin;
  483. } else {
  484. opts.headers.Origin = opts.origin;
  485. }
  486. }
  487. if (parsedUrl.username || parsedUrl.password) {
  488. opts.auth = `${parsedUrl.username}:${parsedUrl.password}`;
  489. }
  490. if (isUnixSocket) {
  491. const parts = opts.path.split(':');
  492. opts.socketPath = parts[0];
  493. opts.path = parts[1];
  494. }
  495. let req = (websocket._req = get(opts));
  496. if (opts.timeout) {
  497. req.on('timeout', () => {
  498. abortHandshake(websocket, req, 'Opening handshake has timed out');
  499. });
  500. }
  501. req.on('error', (err) => {
  502. if (req === null || req.aborted) return;
  503. req = websocket._req = null;
  504. websocket._readyState = WebSocket.CLOSING;
  505. websocket.emit('error', err);
  506. websocket.emitClose();
  507. });
  508. req.on('response', (res) => {
  509. const location = res.headers.location;
  510. const statusCode = res.statusCode;
  511. if (
  512. location &&
  513. opts.followRedirects &&
  514. statusCode >= 300 &&
  515. statusCode < 400
  516. ) {
  517. if (++websocket._redirects > opts.maxRedirects) {
  518. abortHandshake(websocket, req, 'Maximum redirects exceeded');
  519. return;
  520. }
  521. req.abort();
  522. const addr = new URL(location, address);
  523. initAsClient(websocket, addr, protocols, options);
  524. } else if (!websocket.emit('unexpected-response', req, res)) {
  525. abortHandshake(
  526. websocket,
  527. req,
  528. `Unexpected server response: ${res.statusCode}`
  529. );
  530. }
  531. });
  532. req.on('upgrade', (res, socket, head) => {
  533. websocket.emit('upgrade', res);
  534. //
  535. // The user may have closed the connection from a listener of the `upgrade`
  536. // event.
  537. //
  538. if (websocket.readyState !== WebSocket.CONNECTING) return;
  539. req = websocket._req = null;
  540. const digest = createHash('sha1')
  541. .update(key + GUID)
  542. .digest('base64');
  543. if (res.headers['sec-websocket-accept'] !== digest) {
  544. abortHandshake(websocket, socket, 'Invalid Sec-WebSocket-Accept header');
  545. return;
  546. }
  547. const serverProt = res.headers['sec-websocket-protocol'];
  548. const protList = (protocols || '').split(/, */);
  549. let protError;
  550. if (!protocols && serverProt) {
  551. protError = 'Server sent a subprotocol but none was requested';
  552. } else if (protocols && !serverProt) {
  553. protError = 'Server sent no subprotocol';
  554. } else if (serverProt && !protList.includes(serverProt)) {
  555. protError = 'Server sent an invalid subprotocol';
  556. }
  557. if (protError) {
  558. abortHandshake(websocket, socket, protError);
  559. return;
  560. }
  561. if (serverProt) websocket._protocol = serverProt;
  562. if (perMessageDeflate) {
  563. try {
  564. const extensions = parse(res.headers['sec-websocket-extensions']);
  565. if (extensions[PerMessageDeflate.extensionName]) {
  566. perMessageDeflate.accept(extensions[PerMessageDeflate.extensionName]);
  567. websocket._extensions[PerMessageDeflate.extensionName] =
  568. perMessageDeflate;
  569. }
  570. } catch (err) {
  571. abortHandshake(
  572. websocket,
  573. socket,
  574. 'Invalid Sec-WebSocket-Extensions header'
  575. );
  576. return;
  577. }
  578. }
  579. websocket.setSocket(socket, head, opts.maxPayload);
  580. });
  581. }
  582. /**
  583. * Create a `net.Socket` and initiate a connection.
  584. *
  585. * @param {Object} options Connection options
  586. * @return {net.Socket} The newly created socket used to start the connection
  587. * @private
  588. */
  589. function netConnect(options) {
  590. options.path = options.socketPath;
  591. return net.connect(options);
  592. }
  593. /**
  594. * Create a `tls.TLSSocket` and initiate a connection.
  595. *
  596. * @param {Object} options Connection options
  597. * @return {tls.TLSSocket} The newly created socket used to start the connection
  598. * @private
  599. */
  600. function tlsConnect(options) {
  601. options.path = undefined;
  602. if (!options.servername && options.servername !== '') {
  603. options.servername = net.isIP(options.host) ? '' : options.host;
  604. }
  605. return tls.connect(options);
  606. }
  607. /**
  608. * Abort the handshake and emit an error.
  609. *
  610. * @param {WebSocket} websocket The WebSocket instance
  611. * @param {(http.ClientRequest|net.Socket)} stream The request to abort or the
  612. * socket to destroy
  613. * @param {String} message The error message
  614. * @private
  615. */
  616. function abortHandshake(websocket, stream, message) {
  617. websocket._readyState = WebSocket.CLOSING;
  618. const err = new Error(message);
  619. Error.captureStackTrace(err, abortHandshake);
  620. if (stream.setHeader) {
  621. stream.abort();
  622. if (stream.socket && !stream.socket.destroyed) {
  623. //
  624. // On Node.js >= 14.3.0 `request.abort()` does not destroy the socket if
  625. // called after the request completed. See
  626. // https://github.com/websockets/ws/issues/1869.
  627. //
  628. stream.socket.destroy();
  629. }
  630. stream.once('abort', websocket.emitClose.bind(websocket));
  631. websocket.emit('error', err);
  632. } else {
  633. stream.destroy(err);
  634. stream.once('error', websocket.emit.bind(websocket, 'error'));
  635. stream.once('close', websocket.emitClose.bind(websocket));
  636. }
  637. }
  638. /**
  639. * Handle cases where the `ping()`, `pong()`, or `send()` methods are called
  640. * when the `readyState` attribute is `CLOSING` or `CLOSED`.
  641. *
  642. * @param {WebSocket} websocket The WebSocket instance
  643. * @param {*} [data] The data to send
  644. * @param {Function} [cb] Callback
  645. * @private
  646. */
  647. function sendAfterClose(websocket, data, cb) {
  648. if (data) {
  649. const length = toBuffer(data).length;
  650. //
  651. // The `_bufferedAmount` property is used only when the peer is a client and
  652. // the opening handshake fails. Under these circumstances, in fact, the
  653. // `setSocket()` method is not called, so the `_socket` and `_sender`
  654. // properties are set to `null`.
  655. //
  656. if (websocket._socket) websocket._sender._bufferedBytes += length;
  657. else websocket._bufferedAmount += length;
  658. }
  659. if (cb) {
  660. const err = new Error(
  661. `WebSocket is not open: readyState ${websocket.readyState} ` +
  662. `(${readyStates[websocket.readyState]})`
  663. );
  664. cb(err);
  665. }
  666. }
  667. /**
  668. * The listener of the `Receiver` `'conclude'` event.
  669. *
  670. * @param {Number} code The status code
  671. * @param {String} reason The reason for closing
  672. * @private
  673. */
  674. function receiverOnConclude(code, reason) {
  675. const websocket = this[kWebSocket];
  676. websocket._socket.removeListener('data', socketOnData);
  677. websocket._socket.resume();
  678. websocket._closeFrameReceived = true;
  679. websocket._closeMessage = reason;
  680. websocket._closeCode = code;
  681. if (code === 1005) websocket.close();
  682. else websocket.close(code, reason);
  683. }
  684. /**
  685. * The listener of the `Receiver` `'drain'` event.
  686. *
  687. * @private
  688. */
  689. function receiverOnDrain() {
  690. this[kWebSocket]._socket.resume();
  691. }
  692. /**
  693. * The listener of the `Receiver` `'error'` event.
  694. *
  695. * @param {(RangeError|Error)} err The emitted error
  696. * @private
  697. */
  698. function receiverOnError(err) {
  699. const websocket = this[kWebSocket];
  700. websocket._socket.removeListener('data', socketOnData);
  701. websocket._readyState = WebSocket.CLOSING;
  702. websocket._closeCode = err[kStatusCode];
  703. websocket.emit('error', err);
  704. websocket._socket.destroy();
  705. }
  706. /**
  707. * The listener of the `Receiver` `'finish'` event.
  708. *
  709. * @private
  710. */
  711. function receiverOnFinish() {
  712. this[kWebSocket].emitClose();
  713. }
  714. /**
  715. * The listener of the `Receiver` `'message'` event.
  716. *
  717. * @param {(String|Buffer|ArrayBuffer|Buffer[])} data The message
  718. * @private
  719. */
  720. function receiverOnMessage(data) {
  721. this[kWebSocket].emit('message', data);
  722. }
  723. /**
  724. * The listener of the `Receiver` `'ping'` event.
  725. *
  726. * @param {Buffer} data The data included in the ping frame
  727. * @private
  728. */
  729. function receiverOnPing(data) {
  730. const websocket = this[kWebSocket];
  731. websocket.pong(data, !websocket._isServer, NOOP);
  732. websocket.emit('ping', data);
  733. }
  734. /**
  735. * The listener of the `Receiver` `'pong'` event.
  736. *
  737. * @param {Buffer} data The data included in the pong frame
  738. * @private
  739. */
  740. function receiverOnPong(data) {
  741. this[kWebSocket].emit('pong', data);
  742. }
  743. /**
  744. * The listener of the `net.Socket` `'close'` event.
  745. *
  746. * @private
  747. */
  748. function socketOnClose() {
  749. const websocket = this[kWebSocket];
  750. this.removeListener('close', socketOnClose);
  751. this.removeListener('end', socketOnEnd);
  752. websocket._readyState = WebSocket.CLOSING;
  753. //
  754. // The close frame might not have been received or the `'end'` event emitted,
  755. // for example, if the socket was destroyed due to an error. Ensure that the
  756. // `receiver` stream is closed after writing any remaining buffered data to
  757. // it. If the readable side of the socket is in flowing mode then there is no
  758. // buffered data as everything has been already written and `readable.read()`
  759. // will return `null`. If instead, the socket is paused, any possible buffered
  760. // data will be read as a single chunk and emitted synchronously in a single
  761. // `'data'` event.
  762. //
  763. websocket._socket.read();
  764. websocket._receiver.end();
  765. this.removeListener('data', socketOnData);
  766. this[kWebSocket] = undefined;
  767. clearTimeout(websocket._closeTimer);
  768. if (
  769. websocket._receiver._writableState.finished ||
  770. websocket._receiver._writableState.errorEmitted
  771. ) {
  772. websocket.emitClose();
  773. } else {
  774. websocket._receiver.on('error', receiverOnFinish);
  775. websocket._receiver.on('finish', receiverOnFinish);
  776. }
  777. }
  778. /**
  779. * The listener of the `net.Socket` `'data'` event.
  780. *
  781. * @param {Buffer} chunk A chunk of data
  782. * @private
  783. */
  784. function socketOnData(chunk) {
  785. if (!this[kWebSocket]._receiver.write(chunk)) {
  786. this.pause();
  787. }
  788. }
  789. /**
  790. * The listener of the `net.Socket` `'end'` event.
  791. *
  792. * @private
  793. */
  794. function socketOnEnd() {
  795. const websocket = this[kWebSocket];
  796. websocket._readyState = WebSocket.CLOSING;
  797. websocket._receiver.end();
  798. this.end();
  799. }
  800. /**
  801. * The listener of the `net.Socket` `'error'` event.
  802. *
  803. * @private
  804. */
  805. function socketOnError() {
  806. const websocket = this[kWebSocket];
  807. this.removeListener('error', socketOnError);
  808. this.on('error', NOOP);
  809. if (websocket) {
  810. websocket._readyState = WebSocket.CLOSING;
  811. this.destroy();
  812. }
  813. }