From fccc580061a4a35e5f286babafe7416768fd777b Mon Sep 17 00:00:00 2001 From: Luigi Pinca Date: Tue, 12 Dec 2023 08:29:34 +0100 Subject: [PATCH] [fix] Emit the event when the microtask is executed Emit the `'message'`, `'ping'`, and `'pong'` event when the microtask for that event is executed. --- lib/receiver.js | 337 +++++++++++++++++++++++++----------------- test/receiver.test.js | 2 +- 2 files changed, 199 insertions(+), 140 deletions(-) diff --git a/lib/receiver.js b/lib/receiver.js index d0c68432d..18bb9b54d 100644 --- a/lib/receiver.js +++ b/lib/receiver.js @@ -27,7 +27,7 @@ const GET_PAYLOAD_LENGTH_64 = 2; const GET_MASK = 3; const GET_DATA = 4; const INFLATING = 5; -const WAIT_MICROTASK = 6; +const DEFER_EVENT = 6; /** * HyBi Receiver implementation. @@ -78,8 +78,9 @@ class Receiver extends Writable { this._messageLength = 0; this._fragments = []; - this._state = GET_INFO; + this._errored = false; this._loop = false; + this._state = GET_INFO; } /** @@ -151,53 +152,42 @@ class Receiver extends Writable { * @private */ startLoop(cb) { - let err; this._loop = true; do { switch (this._state) { case GET_INFO: - err = this.getInfo(); + this.getInfo(cb); break; case GET_PAYLOAD_LENGTH_16: - err = this.getPayloadLength16(); + this.getPayloadLength16(cb); break; case GET_PAYLOAD_LENGTH_64: - err = this.getPayloadLength64(); + this.getPayloadLength64(cb); break; case GET_MASK: this.getMask(); break; case GET_DATA: - err = this.getData(cb); + this.getData(cb); break; case INFLATING: + case DEFER_EVENT: this._loop = false; return; - default: - // - // `WAIT_MICROTASK`. - // - this._loop = false; - - queueTask(() => { - this._state = GET_INFO; - this.startLoop(cb); - }); - return; } } while (this._loop); - cb(err); + if (!this._errored) cb(); } /** * Reads the first two bytes of a frame. * - * @return {(RangeError|undefined)} A possible error + * @param {Function} cb Callback * @private */ - getInfo() { + getInfo(cb) { if (this._bufferedBytes < 2) { this._loop = false; return; @@ -206,27 +196,31 @@ class Receiver extends Writable { const buf = this.consume(2); if ((buf[0] & 0x30) !== 0x00) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'RSV2 and RSV3 must be clear', true, 1002, 'WS_ERR_UNEXPECTED_RSV_2_3' ); + + cb(error); + return; } const compressed = (buf[0] & 0x40) === 0x40; if (compressed && !this._extensions[PerMessageDeflate.extensionName]) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'RSV1 must be clear', true, 1002, 'WS_ERR_UNEXPECTED_RSV_1' ); + + cb(error); + return; } this._fin = (buf[0] & 0x80) === 0x80; @@ -235,86 +229,100 @@ class Receiver extends Writable { if (this._opcode === 0x00) { if (compressed) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'RSV1 must be clear', true, 1002, 'WS_ERR_UNEXPECTED_RSV_1' ); + + cb(error); + return; } if (!this._fragmented) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'invalid opcode 0', true, 1002, 'WS_ERR_INVALID_OPCODE' ); + + cb(error); + return; } this._opcode = this._fragmented; } else if (this._opcode === 0x01 || this._opcode === 0x02) { if (this._fragmented) { - this._loop = false; - return error( + const error = this.createError( RangeError, `invalid opcode ${this._opcode}`, true, 1002, 'WS_ERR_INVALID_OPCODE' ); + + cb(error); + return; } this._compressed = compressed; } else if (this._opcode > 0x07 && this._opcode < 0x0b) { if (!this._fin) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'FIN must be set', true, 1002, 'WS_ERR_EXPECTED_FIN' ); + + cb(error); + return; } if (compressed) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'RSV1 must be clear', true, 1002, 'WS_ERR_UNEXPECTED_RSV_1' ); + + cb(error); + return; } if ( this._payloadLength > 0x7d || (this._opcode === 0x08 && this._payloadLength === 1) ) { - this._loop = false; - return error( + const error = this.createError( RangeError, `invalid payload length ${this._payloadLength}`, true, 1002, 'WS_ERR_INVALID_CONTROL_PAYLOAD_LENGTH' ); + + cb(error); + return; } } else { - this._loop = false; - return error( + const error = this.createError( RangeError, `invalid opcode ${this._opcode}`, true, 1002, 'WS_ERR_INVALID_OPCODE' ); + + cb(error); + return; } if (!this._fin && !this._fragmented) this._fragmented = this._opcode; @@ -322,54 +330,58 @@ class Receiver extends Writable { if (this._isServer) { if (!this._masked) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'MASK must be set', true, 1002, 'WS_ERR_EXPECTED_MASK' ); + + cb(error); + return; } } else if (this._masked) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'MASK must be clear', true, 1002, 'WS_ERR_UNEXPECTED_MASK' ); + + cb(error); + return; } if (this._payloadLength === 126) this._state = GET_PAYLOAD_LENGTH_16; else if (this._payloadLength === 127) this._state = GET_PAYLOAD_LENGTH_64; - else return this.haveLength(); + else this.haveLength(cb); } /** * Gets extended payload length (7+16). * - * @return {(RangeError|undefined)} A possible error + * @param {Function} cb Callback * @private */ - getPayloadLength16() { + getPayloadLength16(cb) { if (this._bufferedBytes < 2) { this._loop = false; return; } this._payloadLength = this.consume(2).readUInt16BE(0); - return this.haveLength(); + this.haveLength(cb); } /** * Gets extended payload length (7+64). * - * @return {(RangeError|undefined)} A possible error + * @param {Function} cb Callback * @private */ - getPayloadLength64() { + getPayloadLength64(cb) { if (this._bufferedBytes < 8) { this._loop = false; return; @@ -383,38 +395,42 @@ class Receiver extends Writable { // if payload length is greater than this number. // if (num > Math.pow(2, 53 - 32) - 1) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'Unsupported WebSocket frame: payload length > 2^53 - 1', false, 1009, 'WS_ERR_UNSUPPORTED_DATA_PAYLOAD_LENGTH' ); + + cb(error); + return; } this._payloadLength = num * Math.pow(2, 32) + buf.readUInt32BE(4); - return this.haveLength(); + this.haveLength(cb); } /** * Payload length has been read. * - * @return {(RangeError|undefined)} A possible error + * @param {Function} cb Callback * @private */ - haveLength() { + haveLength(cb) { if (this._payloadLength && this._opcode < 0x08) { this._totalPayloadLength += this._payloadLength; if (this._totalPayloadLength > this._maxPayload && this._maxPayload > 0) { - this._loop = false; - return error( + const error = this.createError( RangeError, 'Max payload size exceeded', false, 1009, 'WS_ERR_UNSUPPORTED_MESSAGE_LENGTH' ); + + cb(error); + return; } } @@ -441,7 +457,6 @@ class Receiver extends Writable { * Reads data bytes. * * @param {Function} cb Callback - * @return {(Error|RangeError|undefined)} A possible error * @private */ getData(cb) { @@ -463,7 +478,10 @@ class Receiver extends Writable { } } - if (this._opcode > 0x07) return this.controlMessage(data); + if (this._opcode > 0x07) { + this.controlMessage(data, cb); + return; + } if (this._compressed) { this._state = INFLATING; @@ -480,7 +498,7 @@ class Receiver extends Writable { this._fragments.push(data); } - return this.dataMessage(); + this.dataMessage(cb); } /** @@ -499,76 +517,101 @@ class Receiver extends Writable { if (buf.length) { this._messageLength += buf.length; if (this._messageLength > this._maxPayload && this._maxPayload > 0) { - return cb( - error( - RangeError, - 'Max payload size exceeded', - false, - 1009, - 'WS_ERR_UNSUPPORTED_MESSAGE_LENGTH' - ) + const error = this.createError( + RangeError, + 'Max payload size exceeded', + false, + 1009, + 'WS_ERR_UNSUPPORTED_MESSAGE_LENGTH' ); + + cb(error); + return; } this._fragments.push(buf); } - const er = this.dataMessage(); - if (er) return cb(er); - - this.startLoop(cb); + this.dataMessage(cb); + if (this._state === GET_INFO) this.startLoop(cb); }); } /** * Handles a data message. * - * @return {(Error|undefined)} A possible error + * @param {Function} cb Callback * @private */ - dataMessage() { - if (this._fin) { - const messageLength = this._messageLength; - const fragments = this._fragments; - - this._totalPayloadLength = 0; - this._messageLength = 0; - this._fragmented = 0; - this._fragments = []; - - if (this._opcode === 2) { - let data; - - if (this._binaryType === 'nodebuffer') { - data = concat(fragments, messageLength); - } else if (this._binaryType === 'arraybuffer') { - data = toArrayBuffer(concat(fragments, messageLength)); - } else { - data = fragments; - } + dataMessage(cb) { + if (!this._fin) { + this._state = GET_INFO; + return; + } + + const messageLength = this._messageLength; + const fragments = this._fragments; + + this._totalPayloadLength = 0; + this._messageLength = 0; + this._fragmented = 0; + this._fragments = []; + if (this._opcode === 2) { + let data; + + if (this._binaryType === 'nodebuffer') { + data = concat(fragments, messageLength); + } else if (this._binaryType === 'arraybuffer') { + data = toArrayBuffer(concat(fragments, messageLength)); + } else { + data = fragments; + } + + // + // If the state is `INFLATING`, it means that the frame data was + // decompressed asynchronously, so there is no need to defer the event + // as it will be emitted asynchronously anyway. + // + if (this._state === INFLATING || this._allowMultipleEventsPerMicrotask) { this.emit('message', data, true); + this._state = GET_INFO; } else { - const buf = concat(fragments, messageLength); + this._state = DEFER_EVENT; + queueTask(() => { + this.emit('message', data, true); + this._state = GET_INFO; + this.startLoop(cb); + }); + } + } else { + const buf = concat(fragments, messageLength); - if (!this._skipUTF8Validation && !isValidUTF8(buf)) { - this._loop = false; - return error( - Error, - 'invalid UTF-8 sequence', - true, - 1007, - 'WS_ERR_INVALID_UTF8' - ); - } + if (!this._skipUTF8Validation && !isValidUTF8(buf)) { + const error = this.createError( + Error, + 'invalid UTF-8 sequence', + true, + 1007, + 'WS_ERR_INVALID_UTF8' + ); + + cb(error); + return; + } + if (this._state === INFLATING || this._allowMultipleEventsPerMicrotask) { this.emit('message', buf, false); + this._state = GET_INFO; + } else { + this._state = DEFER_EVENT; + queueTask(() => { + this.emit('message', buf, false); + this._state = GET_INFO; + this.startLoop(cb); + }); } } - - this._state = this._allowMultipleEventsPerMicrotask - ? GET_INFO - : WAIT_MICROTASK; } /** @@ -578,24 +621,26 @@ class Receiver extends Writable { * @return {(Error|RangeError|undefined)} A possible error * @private */ - controlMessage(data) { + controlMessage(data, cb) { if (this._opcode === 0x08) { - this._loop = false; - if (data.length === 0) { + this._loop = false; this.emit('conclude', 1005, EMPTY_BUFFER); this.end(); } else { const code = data.readUInt16BE(0); if (!isValidStatusCode(code)) { - return error( + const error = this.createError( RangeError, `invalid status code ${code}`, true, 1002, 'WS_ERR_INVALID_CLOSE_CODE' ); + + cb(error); + return; } const buf = new FastBuffer( @@ -605,15 +650,19 @@ class Receiver extends Writable { ); if (!this._skipUTF8Validation && !isValidUTF8(buf)) { - return error( + const error = this.createError( Error, 'invalid UTF-8 sequence', true, 1007, 'WS_ERR_INVALID_UTF8' ); + + cb(error); + return; } + this._loop = false; this.emit('conclude', code, buf); this.end(); } @@ -622,38 +671,48 @@ class Receiver extends Writable { return; } - this.emit(this._opcode === 0x09 ? 'ping' : 'pong', data); - this._state = this._allowMultipleEventsPerMicrotask - ? GET_INFO - : WAIT_MICROTASK; + if (this._allowMultipleEventsPerMicrotask) { + this.emit(this._opcode === 0x09 ? 'ping' : 'pong', data); + this._state = GET_INFO; + } else { + this._state = DEFER_EVENT; + queueTask(() => { + this.emit(this._opcode === 0x09 ? 'ping' : 'pong', data); + this._state = GET_INFO; + this.startLoop(cb); + }); + } } -} -module.exports = Receiver; + /** + * Builds an error object. + * + * @param {function(new:Error|RangeError)} ErrorCtor The error constructor + * @param {String} message The error message + * @param {Boolean} prefix Specifies whether or not to add a default prefix to + * `message` + * @param {Number} statusCode The status code + * @param {String} errorCode The exposed error code + * @return {(Error|RangeError)} The error + * @private + */ + createError(ErrorCtor, message, prefix, statusCode, errorCode) { + this._loop = false; + this._errored = true; -/** - * Builds an error object. - * - * @param {function(new:Error|RangeError)} ErrorCtor The error constructor - * @param {String} message The error message - * @param {Boolean} prefix Specifies whether or not to add a default prefix to - * `message` - * @param {Number} statusCode The status code - * @param {String} errorCode The exposed error code - * @return {(Error|RangeError)} The error - * @private - */ -function error(ErrorCtor, message, prefix, statusCode, errorCode) { - const err = new ErrorCtor( - prefix ? `Invalid WebSocket frame: ${message}` : message - ); - - Error.captureStackTrace(err, error); - err.code = errorCode; - err[kStatusCode] = statusCode; - return err; + const err = new ErrorCtor( + prefix ? `Invalid WebSocket frame: ${message}` : message + ); + + Error.captureStackTrace(err, this.createError); + err.code = errorCode; + err[kStatusCode] = statusCode; + return err; + } } +module.exports = Receiver; + /** * A shim for `queueMicrotask()`. * diff --git a/test/receiver.test.js b/test/receiver.test.js index 4e3ee923d..ab2d3c749 100644 --- a/test/receiver.test.js +++ b/test/receiver.test.js @@ -1085,7 +1085,7 @@ describe('Receiver', () => { receiver.write(Buffer.from([0x88, 0x03, 0x03, 0xe8, 0xf8])); }); - it("waits a microtask after the 'message', and 'p{i,o}ng' events", (done) => { + it('emits at most one event per microtask', (done) => { const actual = []; const expected = [ '1',