From 54da52583a1f5b9cc1bd02623f2423f0a0fd6d27 Mon Sep 17 00:00:00 2001
From: Anna Henningsen <anna@addaleax.net>
Date: Sat, 27 Feb 2021 19:46:50 +0100
Subject: [PATCH 1/2] worker: add ports property to MessageEvents
Add `ev.ports` for spec compliancy.
Since we only emit the raw `data` value, and only create the
`MessageEvent` instance if there are EventTarget-style listeners,
we store the ports list temporarily on the MessagePort object itself,
so that we can look it up when we need to create the event object.
Fixes: https://github.com/nodejs/node/issues/37358
---
lib/internal/per_context/messageport.js | 16 +++++--
lib/internal/worker/io.js | 8 +++-
src/node_messaging.cc | 44 +++++++++++++++----
src/node_messaging.h | 12 +++--
.../parallel/test-worker-message-port-move.js | 12 +++--
test/parallel/test-worker-message-port.js | 14 ++++++
6 files changed, 86 insertions(+), 20 deletions(-)
diff --git a/lib/internal/per_context/messageport.js b/lib/internal/per_context/messageport.js
index 43587319b040de..e5df08e6b22368 100644
--- a/lib/internal/per_context/messageport.js
+++ b/lib/internal/per_context/messageport.js
@@ -4,22 +4,30 @@ const {
} = primordials;
class MessageEvent {
- constructor(data, target, type) {
+ constructor(data, target, type, ports) {
this.data = data;
this.target = target;
this.type = type;
+ this.ports = ports ?? [];
}
}
const kHybridDispatch = SymbolFor('nodejs.internal.kHybridDispatch');
+const kCurrentlyReceivingPorts =
+ SymbolFor('nodejs.internal.kCurrentlyReceivingPorts');
-exports.emitMessage = function(data, type) {
+exports.emitMessage = function(data, ports, type) {
if (typeof this[kHybridDispatch] === 'function') {
- this[kHybridDispatch](data, type, undefined);
+ this[kCurrentlyReceivingPorts] = ports;
+ try {
+ this[kHybridDispatch](data, type, undefined);
+ } finally {
+ this[kCurrentlyReceivingPorts] = undefined;
+ }
return;
}
- const event = new MessageEvent(data, this, type);
+ const event = new MessageEvent(data, this, type, ports);
if (type === 'message') {
if (typeof this.onmessage === 'function')
this.onmessage(event);
diff --git a/lib/internal/worker/io.js b/lib/internal/worker/io.js
index aae2dc8a39e585..0c769d261782df 100644
--- a/lib/internal/worker/io.js
+++ b/lib/internal/worker/io.js
@@ -15,6 +15,7 @@ const {
ObjectSetPrototypeOf,
ReflectApply,
Symbol,
+ SymbolFor,
} = primordials;
const {
@@ -70,6 +71,8 @@ const kWritableCallbacks = Symbol('kWritableCallbacks');
const kSource = Symbol('kSource');
const kStartedReading = Symbol('kStartedReading');
const kStdioWantsMoreDataCallback = Symbol('kStdioWantsMoreDataCallback');
+const kCurrentlyReceivingPorts =
+ SymbolFor('nodejs.internal.kCurrentlyReceivingPorts');
const messageTypes = {
UP_AND_RUNNING: 'upAndRunning',
@@ -150,7 +153,9 @@ ObjectDefineProperty(
if (type !== 'message' && type !== 'messageerror') {
return ReflectApply(originalCreateEvent, this, arguments);
}
- return new MessageEvent(type, { data });
+ const ports = this[kCurrentlyReceivingPorts];
+ this[kCurrentlyReceivingPorts] = undefined;
+ return new MessageEvent(type, { data, ports });
},
configurable: false,
writable: false,
@@ -161,6 +166,7 @@ ObjectDefineProperty(
function oninit() {
initNodeEventTarget(this);
setupPortReferencing(this, this, 'message');
+ this[kCurrentlyReceivingPorts] = undefined;
}
defineEventHandler(MessagePort.prototype, 'message');
diff --git a/src/node_messaging.cc b/src/node_messaging.cc
index 2699bd2792e544..62a2064d4d411a 100644
--- a/src/node_messaging.cc
+++ b/src/node_messaging.cc
@@ -126,11 +126,18 @@ class DeserializerDelegate : public ValueDeserializer::Delegate {
} // anonymous namespace
MaybeLocal<Value> Message::Deserialize(Environment* env,
- Local<Context> context) {
+ Local<Context> context,
+ Local<Value>* port_list) {
+ Context::Scope context_scope(context);
+
CHECK(!IsCloseMessage());
+ if (port_list != nullptr && !transferables_.empty()) {
+ // Need to create this outside of the EscapableHandleScope, but inside
+ // the Context::Scope.
+ *port_list = Array::New(env->isolate());
+ }
EscapableHandleScope handle_scope(env->isolate());
- Context::Scope context_scope(context);
// Create all necessary objects for transferables, e.g. MessagePort handles.
std::vector<BaseObjectPtr<BaseObject>> host_objects(transferables_.size());
@@ -146,10 +153,27 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
});
for (uint32_t i = 0; i < transferables_.size(); ++i) {
+ HandleScope handle_scope(env->isolate());
TransferData* data = transferables_[i].get();
host_objects[i] = data->Deserialize(
env, context, std::move(transferables_[i]));
if (!host_objects[i]) return {};
+ if (port_list != nullptr) {
+ // If we gather a list of all message ports, and this transferred object
+ // is a message port, add it to that list. This is a bit of an odd case
+ // of special handling for MessagePorts (as opposed to applying to all
+ // transferables), but it's required for spec compliancy.
+ DCHECK(port_list->IsArray());
+ Local<Array> port_list_array = port_list->As<Array>();
+ Local<Object> obj = host_objects[i]->object();
+ if (env->message_port_constructor_template()->HasInstance(obj)) {
+ if (port_list_array->Set(context,
+ port_list_array->Length(),
+ obj).IsNothing()) {
+ return {};
+ }
+ }
+ }
}
transferables_.clear();
@@ -664,7 +688,8 @@ MessagePort* MessagePort::New(
}
MaybeLocal<Value> MessagePort::ReceiveMessage(Local<Context> context,
- MessageProcessingMode mode) {
+ MessageProcessingMode mode,
+ Local<Value>* port_list) {
std::shared_ptr<Message> received;
{
// Get the head of the message queue.
@@ -696,7 +721,7 @@ MaybeLocal<Value> MessagePort::ReceiveMessage(Local<Context> context,
if (!env()->can_call_into_js()) return MaybeLocal<Value>();
- return received->Deserialize(env(), context);
+ return received->Deserialize(env(), context, port_list);
}
void MessagePort::OnMessage(MessageProcessingMode mode) {
@@ -735,14 +760,15 @@ void MessagePort::OnMessage(MessageProcessingMode mode) {
Local<Function> emit_message = PersistentToLocal::Strong(emit_message_fn_);
Local<Value> payload;
+ Local<Value> port_list = Undefined(env()->isolate());
Local<Value> message_error;
- Local<Value> argv[2];
+ Local<Value> argv[3];
{
// Catch any exceptions from parsing the message itself (not from
// emitting it) as 'messageeror' events.
TryCatchScope try_catch(env());
- if (!ReceiveMessage(context, mode).ToLocal(&payload)) {
+ if (!ReceiveMessage(context, mode, &port_list).ToLocal(&payload)) {
if (try_catch.HasCaught() && !try_catch.HasTerminated())
message_error = try_catch.Exception();
goto reschedule;
@@ -757,13 +783,15 @@ void MessagePort::OnMessage(MessageProcessingMode mode) {
}
argv[0] = payload;
- argv[1] = env()->message_string();
+ argv[1] = port_list;
+ argv[2] = env()->message_string();
if (MakeCallback(emit_message, arraysize(argv), argv).IsEmpty()) {
reschedule:
if (!message_error.IsEmpty()) {
argv[0] = message_error;
- argv[1] = env()->messageerror_string();
+ argv[1] = Undefined(env()->isolate());
+ argv[2] = env()->messageerror_string();
USE(MakeCallback(emit_message, arraysize(argv), argv));
}
diff --git a/src/node_messaging.h b/src/node_messaging.h
index 2e63b22e4ceced..429a5646b8ada4 100644
--- a/src/node_messaging.h
+++ b/src/node_messaging.h
@@ -62,8 +62,10 @@ class Message : public MemoryRetainer {
// Deserialize the contained JS value. May only be called once, and only
// after Serialize() has been called (e.g. by another thread).
- v8::MaybeLocal<v8::Value> Deserialize(Environment* env,
- v8::Local<v8::Context> context);
+ v8::MaybeLocal<v8::Value> Deserialize(
+ Environment* env,
+ v8::Local<v8::Context> context,
+ v8::Local<v8::Value>* port_list = nullptr);
// Serialize a JS value, and optionally transfer objects, into this message.
// The Message object retains ownership of all transferred objects until
@@ -293,8 +295,10 @@ class MessagePort : public HandleWrap {
void OnClose() override;
void OnMessage(MessageProcessingMode mode);
void TriggerAsync();
- v8::MaybeLocal<v8::Value> ReceiveMessage(v8::Local<v8::Context> context,
- MessageProcessingMode mode);
+ v8::MaybeLocal<v8::Value> ReceiveMessage(
+ v8::Local<v8::Context> context,
+ MessageProcessingMode mode,
+ v8::Local<v8::Value>* port_list = nullptr);
std::unique_ptr<MessagePortData> data_ = nullptr;
bool receiving_messages_ = false;
diff --git a/test/parallel/test-worker-message-port-move.js b/test/parallel/test-worker-message-port-move.js
index 7e5d0243fa8b96..44efd2e6a6b94f 100644
--- a/test/parallel/test-worker-message-port-move.js
+++ b/test/parallel/test-worker-message-port-move.js
@@ -34,9 +34,13 @@ vm.runInContext('(' + function() {
assert(!(port instanceof MessagePort));
assert.strictEqual(port.onmessage, undefined);
- port.onmessage = function({ data }) {
+ port.onmessage = function({ data, ports }) {
assert(data instanceof Object);
- port.postMessage(data);
+ assert(ports instanceof Array);
+ assert.strictEqual(ports.length, 1);
+ assert.strictEqual(ports[0], data.p);
+ assert(!(data.p instanceof MessagePort));
+ port.postMessage({});
};
port.start();
}
@@ -55,8 +59,10 @@ vm.runInContext('(' + function() {
}
} + ')()', context);
+const otherChannel = new MessageChannel();
port2.on('message', common.mustCall((msg) => {
assert(msg instanceof Object);
port2.close();
+ otherChannel.port2.close();
}));
-port2.postMessage({});
+port2.postMessage({ p: otherChannel.port1 }, [ otherChannel.port1 ]);
diff --git a/test/parallel/test-worker-message-port.js b/test/parallel/test-worker-message-port.js
index 51618e4fab1850..b5810e726d856c 100644
--- a/test/parallel/test-worker-message-port.js
+++ b/test/parallel/test-worker-message-port.js
@@ -34,6 +34,7 @@ const { MessageChannel, MessagePort } = require('worker_threads');
port1.onmessage = common.mustCall((message) => {
assert.strictEqual(message.data, 4);
assert.strictEqual(message.target, port1);
+ assert.deepStrictEqual(message.ports, []);
port2.close(common.mustCall());
});
@@ -161,6 +162,19 @@ const { MessageChannel, MessagePort } = require('worker_threads');
port1.close();
}
+{
+ // Test MessageEvent#ports
+ const c1 = new MessageChannel();
+ const c2 = new MessageChannel();
+ c1.port1.postMessage({ port: c2.port2 }, [ c2.port2 ]);
+ c1.port2.addEventListener('message', common.mustCall((ev) => {
+ assert.strictEqual(ev.ports.length, 1);
+ assert.strictEqual(ev.ports[0].constructor, MessagePort);
+ c1.port1.close();
+ c2.port1.close();
+ }));
+}
+
{
assert.deepStrictEqual(
Object.getOwnPropertyNames(MessagePort.prototype).sort(),
From 00f5e710c15327a47a471ca5ee21a614f3b75da2 Mon Sep 17 00:00:00 2001
From: Anna Henningsen <anna@addaleax.net>
Date: Mon, 1 Mar 2021 02:27:10 +0100
Subject: [PATCH 2/2] fixup! worker: add ports property to MessageEvents
---
src/node_messaging.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/node_messaging.cc b/src/node_messaging.cc
index 62a2064d4d411a..595f793a6241d4 100644
--- a/src/node_messaging.cc
+++ b/src/node_messaging.cc
@@ -163,7 +163,7 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
// is a message port, add it to that list. This is a bit of an odd case
// of special handling for MessagePorts (as opposed to applying to all
// transferables), but it's required for spec compliancy.
- DCHECK(port_list->IsArray());
+ DCHECK((*port_list)->IsArray());
Local<Array> port_list_array = port_list->As<Array>();
Local<Object> obj = host_objects[i]->object();
if (env->message_port_constructor_template()->HasInstance(obj)) {