Skip to content

Commit

Permalink
[fix] Emit at most one event per event loop iteration
Browse files Browse the repository at this point in the history
Fixes #2216
  • Loading branch information
lpinca committed Apr 11, 2024
1 parent b119b41 commit d40a113
Show file tree
Hide file tree
Showing 3 changed files with 23 additions and 52 deletions.
41 changes: 3 additions & 38 deletions lib/receiver.js
Expand Up @@ -15,12 +15,6 @@ const { isValidStatusCode, isValidUTF8 } = require('./validation');
const FastBuffer = Buffer[Symbol.species];
const promise = Promise.resolve();

Check failure on line 16 in lib/receiver.js

View workflow job for this annotation

GitHub Actions / test (x64, 20, ubuntu-latest)

'promise' is assigned a value but never used

Check failure on line 16 in lib/receiver.js

View workflow job for this annotation

GitHub Actions / test (x64, 20, ubuntu-latest)

'promise' is assigned a value but never used

//
// `queueMicrotask()` is not available in Node.js < 11.
//
const queueTask =
typeof queueMicrotask === 'function' ? queueMicrotask : queueMicrotaskShim;

const GET_INFO = 0;
const GET_PAYLOAD_LENGTH_16 = 1;
const GET_PAYLOAD_LENGTH_64 = 2;
Expand Down Expand Up @@ -577,7 +571,7 @@ class Receiver extends Writable {
this._state = GET_INFO;
} else {
this._state = DEFER_EVENT;
queueTask(() => {
setImmediate(() => {
this.emit('message', data, true);
this._state = GET_INFO;
this.startLoop(cb);
Expand All @@ -604,7 +598,7 @@ class Receiver extends Writable {
this._state = GET_INFO;
} else {
this._state = DEFER_EVENT;
queueTask(() => {
setImmediate(() => {
this.emit('message', buf, false);
this._state = GET_INFO;
this.startLoop(cb);
Expand Down Expand Up @@ -675,7 +669,7 @@ class Receiver extends Writable {
this._state = GET_INFO;
} else {
this._state = DEFER_EVENT;
queueTask(() => {
setImmediate(() => {
this.emit(this._opcode === 0x09 ? 'ping' : 'pong', data);
this._state = GET_INFO;
this.startLoop(cb);
Expand Down Expand Up @@ -711,32 +705,3 @@ class Receiver extends Writable {
}

module.exports = Receiver;

/**
* A shim for `queueMicrotask()`.
*
* @param {Function} cb Callback
*/
function queueMicrotaskShim(cb) {
promise.then(cb).catch(throwErrorNextTick);
}

/**
* Throws an error.
*
* @param {Error} err The error to throw
* @private
*/
function throwError(err) {
throw err;
}

/**
* Throws an error in the next tick.
*
* @param {Error} err The error to throw
* @private
*/
function throwErrorNextTick(err) {
process.nextTick(throwError, err);
}
28 changes: 18 additions & 10 deletions test/receiver.test.js
Expand Up @@ -1085,17 +1085,21 @@ describe('Receiver', () => {
receiver.write(Buffer.from([0x88, 0x03, 0x03, 0xe8, 0xf8]));
});

it('emits at most one event per microtask', (done) => {
it('emits at most one event per event loop iteration', (done) => {
const actual = [];
const expected = [
'1',
'microtask 1',
'- 1',
'-- 1',
'2',
'microtask 2',
'- 2',
'-- 2',
'3',
'microtask 3',
'- 3',
'-- 3',
'4',
'microtask 4'
'- 4',
'-- 4'
];

function listener(data) {
Expand All @@ -1104,12 +1108,16 @@ describe('Receiver', () => {

// `queueMicrotask()` is not available in Node.js < 11.
Promise.resolve().then(() => {
actual.push(`microtask ${message}`);
actual.push(`- ${message}`);

if (actual.length === 8) {
assert.deepStrictEqual(actual, expected);
done();
}
Promise.resolve().then(() => {
actual.push(`-- ${message}`);

if (actual.length === 12) {
assert.deepStrictEqual(actual, expected);
done();
}
});
});
}

Expand Down
6 changes: 2 additions & 4 deletions test/websocket.test.js
Expand Up @@ -4205,8 +4205,7 @@ describe('WebSocket', () => {

if (messages.push(message.toString()) > 1) return;

// `queueMicrotask()` is not available in Node.js < 11.
Promise.resolve().then(() => {
setImmediate(() => {
process.nextTick(() => {
assert.strictEqual(ws._receiver._state, 5);
ws.close(1000);
Expand Down Expand Up @@ -4456,8 +4455,7 @@ describe('WebSocket', () => {

if (messages.push(message.toString()) > 1) return;

// `queueMicrotask()` is not available in Node.js < 11.
Promise.resolve().then(() => {
setImmediate(() => {
process.nextTick(() => {
assert.strictEqual(ws._receiver._state, 5);
ws.terminate();
Expand Down

0 comments on commit d40a113

Please sign in to comment.