GeezerVison is showing github.com — worker: add ports property to MessageEvents · nodejs/node@8024ffb
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: lib/internal/per_context/messageport.js +12-4Lines changed: 12 additions & 4 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -4,22 +4,30 @@ const { | 4 | 4 | } = primordials; | 5 | 5 |
| 6 | 6 | class MessageEvent { | 7 | | - constructor(data, target, type) { | | 7 | + constructor(data, target, type, ports) { | 8 | 8 | this.data = data; | 9 | 9 | this.target = target; | 10 | 10 | this.type = type; | | 11 | + this.ports = ports ?? []; | 11 | 12 | } | 12 | 13 | } | 13 | 14 |
| 14 | 15 | const kHybridDispatch = SymbolFor('nodejs.internal.kHybridDispatch'); | | 16 | +const kCurrentlyReceivingPorts = | | 17 | + SymbolFor('nodejs.internal.kCurrentlyReceivingPorts'); | 15 | 18 |
| 16 | | -exports.emitMessage = function(data, type) { | | 19 | +exports.emitMessage = function(data, ports, type) { | 17 | 20 | if (typeof this[kHybridDispatch] === 'function') { | 18 | | - this[kHybridDispatch](data, type, undefined); | | 21 | + this[kCurrentlyReceivingPorts] = ports; | | 22 | + try { | | 23 | + this[kHybridDispatch](data, type, undefined); | | 24 | + } finally { | | 25 | + this[kCurrentlyReceivingPorts] = undefined; | | 26 | + } | 19 | 27 | return; | 20 | 28 | } | 21 | 29 |
| 22 | | - const event = new MessageEvent(data, this, type); | | 30 | + const event = new MessageEvent(data, this, type, ports); | 23 | 31 | if (type === 'message') { | 24 | 32 | if (typeof this.onmessage === 'function') | 25 | 33 | this.onmessage(event); | |
Collapse file Copy file name to clipboardExpand all lines: lib/internal/worker/io.js +7-1Lines changed: 7 additions & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -15,6 +15,7 @@ const { | 15 | 15 | ObjectSetPrototypeOf, | 16 | 16 | ReflectApply, | 17 | 17 | Symbol, | | 18 | + SymbolFor, | 18 | 19 | } = primordials; | 19 | 20 |
| 20 | 21 | const { | @@ -70,6 +71,8 @@ const kWritableCallbacks = Symbol('kWritableCallbacks'); | 70 | 71 | const kSource = Symbol('kSource'); | 71 | 72 | const kStartedReading = Symbol('kStartedReading'); | 72 | 73 | const kStdioWantsMoreDataCallback = Symbol('kStdioWantsMoreDataCallback'); | | 74 | +const kCurrentlyReceivingPorts = | | 75 | + SymbolFor('nodejs.internal.kCurrentlyReceivingPorts'); | 73 | 76 |
| 74 | 77 | const messageTypes = { | 75 | 78 | UP_AND_RUNNING: 'upAndRunning', | @@ -150,7 +153,9 @@ ObjectDefineProperty( | 150 | 153 | if (type !== 'message' && type !== 'messageerror') { | 151 | 154 | return ReflectApply(originalCreateEvent, this, arguments); | 152 | 155 | } | 153 | | - return new MessageEvent(type, { data }); | | 156 | + const ports = this[kCurrentlyReceivingPorts]; | | 157 | + this[kCurrentlyReceivingPorts] = undefined; | | 158 | + return new MessageEvent(type, { data, ports }); | 154 | 159 | }, | 155 | 160 | configurable: false, | 156 | 161 | writable: false, | @@ -161,6 +166,7 @@ ObjectDefineProperty( | 161 | 166 | function oninit() { | 162 | 167 | initNodeEventTarget(this); | 163 | 168 | setupPortReferencing(this, this, 'message'); | | 169 | + this[kCurrentlyReceivingPorts] = undefined; | 164 | 170 | } | 165 | 171 |
| 166 | 172 | defineEventHandler(MessagePort.prototype, 'message'); | |
Collapse file Copy file name to clipboardExpand all lines: src/node_messaging.cc +36-8Lines changed: 36 additions & 8 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -126,11 +126,18 @@ class DeserializerDelegate : public ValueDeserializer::Delegate { | 126 | 126 | } // anonymous namespace | 127 | 127 |
| 128 | 128 | MaybeLocal<Value> Message::Deserialize(Environment* env, | 129 | | - Local<Context> context) { | | 129 | + Local<Context> context, | | 130 | + Local<Value>* port_list) { | | 131 | + Context::Scope context_scope(context); | | 132 | + | 130 | 133 | CHECK(!IsCloseMessage()); | | 134 | + if (port_list != nullptr && !transferables_.empty()) { | | 135 | + // Need to create this outside of the EscapableHandleScope, but inside | | 136 | + // the Context::Scope. | | 137 | + *port_list = Array::New(env->isolate()); | | 138 | + } | 131 | 139 |
| 132 | 140 | EscapableHandleScope handle_scope(env->isolate()); | 133 | | - Context::Scope context_scope(context); | 134 | 141 |
| 135 | 142 | // Create all necessary objects for transferables, e.g. MessagePort handles. | 136 | 143 | std::vector<BaseObjectPtr<BaseObject>> host_objects(transferables_.size()); | @@ -146,10 +153,27 @@ MaybeLocal<Value> Message::Deserialize(Environment* env, | 146 | 153 | }); | 147 | 154 |
| 148 | 155 | for (uint32_t i = 0; i < transferables_.size(); ++i) { | | 156 | + HandleScope handle_scope(env->isolate()); | 149 | 157 | TransferData* data = transferables_[i].get(); | 150 | 158 | host_objects[i] = data->Deserialize( | 151 | 159 | env, context, std::move(transferables_[i])); | 152 | 160 | if (!host_objects[i]) return {}; | | 161 | + if (port_list != nullptr) { | | 162 | + // If we gather a list of all message ports, and this transferred object | | 163 | + // is a message port, add it to that list. This is a bit of an odd case | | 164 | + // of special handling for MessagePorts (as opposed to applying to all | | 165 | + // transferables), but it's required for spec compliancy. | | 166 | + DCHECK((*port_list)->IsArray()); | | 167 | + Local<Array> port_list_array = port_list->As<Array>(); | | 168 | + Local<Object> obj = host_objects[i]->object(); | | 169 | + if (env->message_port_constructor_template()->HasInstance(obj)) { | | 170 | + if (port_list_array->Set(context, | | 171 | + port_list_array->Length(), | | 172 | + obj).IsNothing()) { | | 173 | + return {}; | | 174 | + } | | 175 | + } | | 176 | + } | 153 | 177 | } | 154 | 178 | transferables_.clear(); | 155 | 179 |
| @@ -664,7 +688,8 @@ MessagePort* MessagePort::New( | 664 | 688 | } | 665 | 689 |
| 666 | 690 | MaybeLocal<Value> MessagePort::ReceiveMessage(Local<Context> context, | 667 | | - MessageProcessingMode mode) { | | 691 | + MessageProcessingMode mode, | | 692 | + Local<Value>* port_list) { | 668 | 693 | std::shared_ptr<Message> received; | 669 | 694 | { | 670 | 695 | // Get the head of the message queue. | @@ -696,7 +721,7 @@ MaybeLocal<Value> MessagePort::ReceiveMessage(Local<Context> context, | 696 | 721 |
| 697 | 722 | if (!env()->can_call_into_js()) return MaybeLocal<Value>(); | 698 | 723 |
| 699 | | - return received->Deserialize(env(), context); | | 724 | + return received->Deserialize(env(), context, port_list); | 700 | 725 | } | 701 | 726 |
| 702 | 727 | void MessagePort::OnMessage(MessageProcessingMode mode) { | @@ -735,14 +760,15 @@ void MessagePort::OnMessage(MessageProcessingMode mode) { | 735 | 760 | Local<Function> emit_message = PersistentToLocal::Strong(emit_message_fn_); | 736 | 761 |
| 737 | 762 | Local<Value> payload; | | 763 | + Local<Value> port_list = Undefined(env()->isolate()); | 738 | 764 | Local<Value> message_error; | 739 | | - Local<Value> argv[2]; | | 765 | + Local<Value> argv[3]; | 740 | 766 |
| 741 | 767 | { | 742 | 768 | // Catch any exceptions from parsing the message itself (not from | 743 | 769 | // emitting it) as 'messageeror' events. | 744 | 770 | TryCatchScope try_catch(env()); | 745 | | - if (!ReceiveMessage(context, mode).ToLocal(&payload)) { | | 771 | + if (!ReceiveMessage(context, mode, &port_list).ToLocal(&payload)) { | 746 | 772 | if (try_catch.HasCaught() && !try_catch.HasTerminated()) | 747 | 773 | message_error = try_catch.Exception(); | 748 | 774 | goto reschedule; | @@ -757,13 +783,15 @@ void MessagePort::OnMessage(MessageProcessingMode mode) { | 757 | 783 | } | 758 | 784 |
| 759 | 785 | argv[0] = payload; | 760 | | - argv[1] = env()->message_string(); | | 786 | + argv[1] = port_list; | | 787 | + argv[2] = env()->message_string(); | 761 | 788 |
| 762 | 789 | if (MakeCallback(emit_message, arraysize(argv), argv).IsEmpty()) { | 763 | 790 | reschedule: | 764 | 791 | if (!message_error.IsEmpty()) { | 765 | 792 | argv[0] = message_error; | 766 | | - argv[1] = env()->messageerror_string(); | | 793 | + argv[1] = Undefined(env()->isolate()); | | 794 | + argv[2] = env()->messageerror_string(); | 767 | 795 | USE(MakeCallback(emit_message, arraysize(argv), argv)); | 768 | 796 | } | 769 | 797 |
| |
Collapse file Copy file name to clipboardExpand all lines: src/node_messaging.h +8-4Lines changed: 8 additions & 4 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -62,8 +62,10 @@ class Message : public MemoryRetainer { | 62 | 62 |
| 63 | 63 | // Deserialize the contained JS value. May only be called once, and only | 64 | 64 | // after Serialize() has been called (e.g. by another thread). | 65 | | - v8::MaybeLocal<v8::Value> Deserialize(Environment* env, | 66 | | - v8::Local<v8::Context> context); | | 65 | + v8::MaybeLocal<v8::Value> Deserialize( | | 66 | + Environment* env, | | 67 | + v8::Local<v8::Context> context, | | 68 | + v8::Local<v8::Value>* port_list = nullptr); | 67 | 69 |
| 68 | 70 | // Serialize a JS value, and optionally transfer objects, into this message. | 69 | 71 | // The Message object retains ownership of all transferred objects until | @@ -293,8 +295,10 @@ class MessagePort : public HandleWrap { | 293 | 295 | void OnClose() override; | 294 | 296 | void OnMessage(MessageProcessingMode mode); | 295 | 297 | void TriggerAsync(); | 296 | | - v8::MaybeLocal<v8::Value> ReceiveMessage(v8::Local<v8::Context> context, | 297 | | - MessageProcessingMode mode); | | 298 | + v8::MaybeLocal<v8::Value> ReceiveMessage( | | 299 | + v8::Local<v8::Context> context, | | 300 | + MessageProcessingMode mode, | | 301 | + v8::Local<v8::Value>* port_list = nullptr); | 298 | 302 |
| 299 | 303 | std::unique_ptr<MessagePortData> data_ = nullptr; | 300 | 304 | bool receiving_messages_ = false; | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-worker-message-port-move.js +9-3Lines changed: 9 additions & 3 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -34,9 +34,13 @@ vm.runInContext('(' + function() { | 34 | 34 |
| 35 | 35 | assert(!(port instanceof MessagePort)); | 36 | 36 | assert.strictEqual(port.onmessage, undefined); | 37 | | - port.onmessage = function({ data }) { | | 37 | + port.onmessage = function({ data, ports }) { | 38 | 38 | assert(data instanceof Object); | 39 | | - port.postMessage(data); | | 39 | + assert(ports instanceof Array); | | 40 | + assert.strictEqual(ports.length, 1); | | 41 | + assert.strictEqual(ports[0], data.p); | | 42 | + assert(!(data.p instanceof MessagePort)); | | 43 | + port.postMessage({}); | 40 | 44 | }; | 41 | 45 | port.start(); | 42 | 46 | } | @@ -55,8 +59,10 @@ vm.runInContext('(' + function() { | 55 | 59 | } | 56 | 60 | } + ')()', context); | 57 | 61 |
| | 62 | +const otherChannel = new MessageChannel(); | 58 | 63 | port2.on('message', common.mustCall((msg) => { | 59 | 64 | assert(msg instanceof Object); | 60 | 65 | port2.close(); | | 66 | + otherChannel.port2.close(); | 61 | 67 | })); | 62 | | -port2.postMessage({}); | | 68 | +port2.postMessage({ p: otherChannel.port1 }, [ otherChannel.port1 ]); |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-worker-message-port.js +14Lines changed: 14 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -34,6 +34,7 @@ const { MessageChannel, MessagePort } = require('worker_threads'); | 34 | 34 | port1.onmessage = common.mustCall((message) => { | 35 | 35 | assert.strictEqual(message.data, 4); | 36 | 36 | assert.strictEqual(message.target, port1); | | 37 | + assert.deepStrictEqual(message.ports, []); | 37 | 38 | port2.close(common.mustCall()); | 38 | 39 | }); | 39 | 40 |
| @@ -161,6 +162,19 @@ const { MessageChannel, MessagePort } = require('worker_threads'); | 161 | 162 | port1.close(); | 162 | 163 | } | 163 | 164 |
| | 165 | +{ | | 166 | + // Test MessageEvent#ports | | 167 | + const c1 = new MessageChannel(); | | 168 | + const c2 = new MessageChannel(); | | 169 | + c1.port1.postMessage({ port: c2.port2 }, [ c2.port2 ]); | | 170 | + c1.port2.addEventListener('message', common.mustCall((ev) => { | | 171 | + assert.strictEqual(ev.ports.length, 1); | | 172 | + assert.strictEqual(ev.ports[0].constructor, MessagePort); | | 173 | + c1.port1.close(); | | 174 | + c2.port1.close(); | | 175 | + })); | | 176 | +} | | 177 | + | 164 | 178 | { | 165 | 179 | assert.deepStrictEqual( | 166 | 180 | Object.getOwnPropertyNames(MessagePort.prototype).sort(), | |
|