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: #37358
PR-URL: #37538
Reviewed-By: Michaël Zasso <targos@protonmail.com>
Reviewed-By: Joyee Cheung <joyeec9h3@gmail.com>

Author: addaleax

lib/internal/per_context/messageport.js

@@ -4,22 +4,30 @@ const {
 }=primordials;
 
 classMessageEvent{
-constructor(data,target,type){
+constructor(data,target,type,ports){
 this.data=data;
 this.target=target;
 this.type=type;
+this.ports=ports??[];
 }
 }
 
 constkHybridDispatch=SymbolFor('nodejs.internal.kHybridDispatch');
+constkCurrentlyReceivingPorts=
+SymbolFor('nodejs.internal.kCurrentlyReceivingPorts');
 
-exports.emitMessage=function(data,type){
+exports.emitMessage=function(data,ports,type){
 if(typeofthis[kHybridDispatch]==='function'){
-this[kHybridDispatch](data,type,undefined);
+this[kCurrentlyReceivingPorts]=ports;
+try{
+this[kHybridDispatch](data,type,undefined);
+}finally{
+this[kCurrentlyReceivingPorts]=undefined;
+}
 return;
 }
 
-constevent=newMessageEvent(data,this,type);
+constevent=newMessageEvent(data,this,type,ports);
 if(type==='message'){
 if(typeofthis.onmessage==='function')
 this.onmessage(event);

lib/internal/worker/io.js

@@ -15,6 +15,7 @@ const {
  ObjectSetPrototypeOf,
  ReflectApply,
  Symbol,
+ SymbolFor,
 }=primordials;
 
 const{
@@ -70,6 +71,8 @@ const kWritableCallbacks = Symbol('kWritableCallbacks');
 constkSource=Symbol('kSource');
 constkStartedReading=Symbol('kStartedReading');
 constkStdioWantsMoreDataCallback=Symbol('kStdioWantsMoreDataCallback');
+constkCurrentlyReceivingPorts=
+SymbolFor('nodejs.internal.kCurrentlyReceivingPorts');
 
 constmessageTypes={
 UP_AND_RUNNING: 'upAndRunning',
@@ -150,7 +153,9 @@ ObjectDefineProperty(
 if(type!=='message'&&type!=='messageerror'){
 returnReflectApply(originalCreateEvent,this,arguments);
 }
-returnnewMessageEvent(type,{ data });
+constports=this[kCurrentlyReceivingPorts];
+this[kCurrentlyReceivingPorts]=undefined;
+returnnewMessageEvent(type,{ data, ports });
 },
 configurable: false,
 writable: false,
@@ -161,6 +166,7 @@ ObjectDefineProperty(
 functiononinit(){
 initNodeEventTarget(this);
 setupPortReferencing(this,this,'message');
+this[kCurrentlyReceivingPorts]=undefined;
 }
 
 defineEventHandler(MessagePort.prototype,'message');

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);
 }
 
 voidMessagePort::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));
  }
 

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 {
 voidOnClose() override;
 voidOnMessage(MessageProcessingMode mode);
 voidTriggerAsync();
- 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;

test/parallel/test-worker-message-port-move.js

@@ -34,9 +34,13 @@ vm.runInContext('(' + function() {
 
 assert(!(portinstanceofMessagePort));
 assert.strictEqual(port.onmessage,undefined);
-port.onmessage=function({ data }){
+port.onmessage=function({ data, ports}){
 assert(datainstanceofObject);
-port.postMessage(data);
+assert(portsinstanceofArray);
+assert.strictEqual(ports.length,1);
+assert.strictEqual(ports[0],data.p);
+assert(!(data.pinstanceofMessagePort));
+port.postMessage({});
 };
 port.start();
 }
@@ -55,8 +59,10 @@ vm.runInContext('(' + function() {
 }
 }+')()',context);
 
+constotherChannel=newMessageChannel();
 port2.on('message',common.mustCall((msg)=>{
 assert(msginstanceofObject);
 port2.close();
+otherChannel.port2.close();
 }));
-port2.postMessage({});
+port2.postMessage({p: otherChannel.port1},[otherChannel.port1]);

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
+constc1=newMessageChannel();
+constc2=newMessageChannel();
+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(),