diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 5fa6aa8707f8..5d23baf6825f 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -128,25 +128,26 @@ function injectFakeEmitter(Class) { return event.detail; } - function wrapped(run, listener) { + // node-style listeners run with `this` = the object on()/once() was called on. + function wrapped(target, run, listener) { return function (event) { - return listener(run(event)); + return listener.$call(target, run(event)); }; } - function functionForEventType(event, listener) { + function functionForEventType(target, event, listener) { switch (event) { case "error": case "messageerror": { - return wrapped(errorEventHandler, listener); + return wrapped(target, errorEventHandler, listener); } case "message": { - return wrapped(messageEventHandler, listener); + return wrapped(target, messageEventHandler, listener); } default: { - return wrapped(customEventHandler, listener); + return wrapped(target, customEventHandler, listener); } } } @@ -173,7 +174,7 @@ function injectFakeEmitter(Class) { } function on(event, listener) { - register(this, event, listener, functionForEventType(event, listener), undefined); + register(this, event, listener, functionForEventType(this, event, listener), undefined); return this; } @@ -190,8 +191,8 @@ function injectFakeEmitter(Class) { } function once(event, listener) { - const wrapper = functionForEventType(event, listener); const target = this; + const wrapper = functionForEventType(target, event, listener); // EventTarget drops a {once:true} listener natively, without telling the // registry — so purge it here or listenerCount()/eventNames() keep counting // a listener that already fired. diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 023ea833ffa1..d48fcf76432c 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -1418,6 +1418,132 @@ test("off() removes a pending once() listener", () => { port2.close(); }); +// node invokes node-style listeners with `this` bound to the port. The on()/once() +// wrappers used to call the user's function bare, so `this` was undefined. Covers +// every way a wrapper gets invoked: a native MessageEvent, a native close Event +// (on, once and close(cb), which is once("close", cb)), and emit(). The last +// listener registered for each event is the one that resolves, so every earlier +// one has already run by the time the promise settles. +test("on()/once() listeners are called with `this` bound to the port", async () => { + const { port1, port2 } = new MessageChannel(); + const calls: unknown[] = []; + const { promise: messaged, resolve: onMessaged } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + + port1.on("message", function (this: unknown, data) { + calls.push(["on message", this === port1, data]); + }); + port1.once("message", function (this: unknown, data) { + calls.push(["once message", this === port1, data]); + onMessaged(); + }); + port1.on("close", function (this: unknown) { + calls.push(["on close", this === port1]); + }); + port1.once("close", function (this: unknown) { + calls.push(["once close", this === port1]); + }); + port1.on("custom", function (this: unknown, arg) { + calls.push(["on custom", this === port1, arg]); + }); + port1.once("custom", function (this: unknown, arg) { + calls.push(["once custom", this === port1, arg]); + }); + + port1.emit("custom", 42); + port2.postMessage("hi"); + await messaged; + port1.close(function (this: unknown) { + calls.push(["close(cb)", this === port1]); + onClosed(); + }); + await closed; + port2.close(); + + expect(calls).toEqual([ + ["on custom", true, 42], + ["once custom", true, 42], + ["on message", true, "hi"], + ["once message", true, "hi"], + ["on close", true], + ["once close", true], + ["close(cb)", true], + ]); +}); + +test("parentPort.on() listeners are called with `this` bound to parentPort", async () => { + const w = new Worker( + `const { parentPort } = require("node:worker_threads"); + parentPort.once("message", function (data) { + this.postMessage({ data, thisIsParentPort: this === parentPort }); + });`, + { eval: true }, + ); + w.postMessage("ping"); + const [reply] = await once(w, "message"); + await w.terminate(); + expect(reply).toEqual({ data: "ping", thisIsParentPort: true }); +}); + +// In a plain globalThis.Worker, parentPort is a stand-in object whose +// addEventListener forwards to the worker's global scope, so EventTarget invokes +// the wrapper with the global scope as `this`. The listener must still see the +// object it was registered on. +test("the stand-in parentPort of a non-node Worker also binds `this` to parentPort", async () => { + const url = URL.createObjectURL( + new Blob([ + `const { parentPort } = require("worker_threads"); + parentPort.on("message", function (data) { + self.postMessage({ data, thisIsParentPort: this === parentPort, thisIsGlobal: this === self }); + });`, + ]), + ); + const w = new globalThis.Worker(url); + try { + const { promise, resolve, reject } = Promise.withResolvers(); + w.onmessage = e => resolve(e.data); + w.onerror = e => reject(new Error(e.message)); + w.postMessage("ping"); + expect(await promise).toEqual({ data: "ping", thisIsParentPort: true, thisIsGlobal: false }); + } finally { + w.terminate(); + URL.revokeObjectURL(url); + } +}); + +// The wrappers hand the listener's return value back to EventTarget, which is +// how a rejecting async on()/once() listener becomes an uncaughtException, as in +// node. Spawned because the test runner would otherwise see the exceptions itself. +test("a rejecting async on()/once() listener is reported as an uncaughtException", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const { MessageChannel } = require("worker_threads"); + const { port1, port2 } = new MessageChannel(); + const seen = []; + process.on("uncaughtException", err => { + seen.push(err.message); + if (seen.length === 2) { + console.log(JSON.stringify(seen)); + port1.close(); + } + }); + port1.on("message", async () => { throw new Error("from on"); }); + port1.once("message", async () => { throw new Error("from once"); }); + port2.postMessage(1);`, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ + stdout: JSON.stringify(["from on", "from once"]), + stderr: "", + exitCode: 0, + }); +}); + test("close(cb) interleaves with other close listeners in registration order", async () => { // node's mechanism is `this.once('close', cb)`, so cb interleaves with other // close listeners in the order they were registered (verified against node).