Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit f447acd

Browse filesBrowse files
addaleaxtargos
addaleax
authored and
targos
committed
worker: support MessagePort passing in messages
Support passing `MessagePort` instances through other `MessagePort`s, as expected by the `MessagePort` spec. Thanks to Stephen Belanger for reviewing this change in its original PR. Refs: ayojs/ayo#106 PR-URL: #20876 Reviewed-By: Gireesh Punathil <gpunathi@in.ibm.com> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Shingo Inoue <leko.noor@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Tiancheng "Timothy" Gu <timothygu99@gmail.com> Reviewed-By: John-David Dalton <john.david.dalton@gmail.com> Reviewed-By: Gus Caplan <me@gus.host>
1 parent 337be58 commit f447acd
Copy full SHA for f447acd

File tree

Expand file treeCollapse file tree

6 files changed

+123
-6
lines changed
Open diff view settings
Filter options
Expand file treeCollapse file tree

6 files changed

+123
-6
lines changed
Open diff view settings
Collapse file

‎doc/api/errors.md‎

Copy file name to clipboardExpand all lines: doc/api/errors.md
+12Lines changed: 12 additions & 0 deletions
  • Display the source diff
  • Display the rich diff
Original file line numberDiff line numberDiff line change
@@ -629,6 +629,12 @@ An operation outside the bounds of a `Buffer` was attempted.
629629
An attempt has been made to create a `Buffer` larger than the maximum allowed
630630
size.
631631

632+
<a id="ERR_CANNOT_TRANSFER_OBJECT"></a>
633+
### ERR_CANNOT_TRANSFER_OBJECT
634+
635+
The value passed to `postMessage()` contained an object that is not supported
636+
for transferring.
637+
632638
<a id="ERR_CANNOT_WATCH_SIGINT"></a>
633639
### ERR_CANNOT_WATCH_SIGINT
634640

@@ -1304,6 +1310,12 @@ strict compliance with the API specification (which in some cases may accept
13041310
`func(undefined)` and `func()` are treated identically, and the
13051311
[`ERR_INVALID_ARG_TYPE`][] error code may be used instead.
13061312

1313+
<a id="ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST"></a>
1314+
### ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST
1315+
1316+
A `MessagePort` was found in the object passed to a `postMessage()` call,
1317+
but not provided in the `transferList` for that call.
1318+
13071319
<a id="ERR_MISSING_MODULE"></a>
13081320
### ERR_MISSING_MODULE
13091321

Collapse file

‎doc/api/worker.md‎

Copy file name to clipboardExpand all lines: doc/api/worker.md
+1-1Lines changed: 1 addition & 1 deletion
  • Display the source diff
  • Display the rich diff
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ the [HTML structured clone algorithm][]. In particular, it may contain circular
8383
references and objects like typed arrays that the `JSON` API is not able
8484
to stringify.
8585

86-
`transferList` may be a list of `ArrayBuffer` objects.
86+
`transferList` may be a list of `ArrayBuffer` and `MessagePort` objects.
8787
After transferring, they will not be usable on the sending side of the channel
8888
anymore (even if they are not contained in `value`).
8989

Collapse file

‎src/node_errors.h‎

Copy file name to clipboardExpand all lines: src/node_errors.h
+6-1Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ namespace node {
1919
#define ERRORS_WITH_CODE(V) \
2020
V(ERR_BUFFER_OUT_OF_BOUNDS, RangeError) \
2121
V(ERR_BUFFER_TOO_LARGE, Error) \
22+
V(ERR_CANNOT_TRANSFER_OBJECT, TypeError) \
2223
V(ERR_CLOSED_MESSAGE_PORT, Error) \
2324
V(ERR_CONSTRUCT_CALL_REQUIRED, Error) \
2425
V(ERR_INDEX_OUT_OF_RANGE, RangeError) \
@@ -27,6 +28,7 @@ namespace node {
2728
V(ERR_INVALID_TRANSFER_OBJECT, TypeError) \
2829
V(ERR_MEMORY_ALLOCATION_FAILED, Error) \
2930
V(ERR_MISSING_ARGS, TypeError) \
31+
V(ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST, TypeError) \
3032
V(ERR_MISSING_MODULE, Error) \
3133
V(ERR_STRING_TOO_LONG, Error) \
3234

@@ -51,11 +53,14 @@ namespace node {
5153
// Errors with predefined static messages
5254

5355
#define PREDEFINED_ERROR_MESSAGES(V) \
56+
V(ERR_CANNOT_TRANSFER_OBJECT, "Cannot transfer object of unsupported type")\
5457
V(ERR_CLOSED_MESSAGE_PORT, "Cannot send data on closed MessagePort") \
5558
V(ERR_CONSTRUCT_CALL_REQUIRED, "Cannot call constructor without `new`") \
5659
V(ERR_INDEX_OUT_OF_RANGE, "Index out of range") \
5760
V(ERR_INVALID_TRANSFER_OBJECT, "Found invalid object in transferList") \
58-
V(ERR_MEMORY_ALLOCATION_FAILED, "Failed to allocate memory")
61+
V(ERR_MEMORY_ALLOCATION_FAILED, "Failed to allocate memory") \
62+
V(ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST, \
63+
"MessagePort was found in message but not listed in transferList")
5964

6065
#define V(code, message) \
6166
inline v8::Local<v8::Value> code(v8::Isolate* isolate) { \
Collapse file

‎src/node_messaging.cc‎

Copy file name to clipboardExpand all lines: src/node_messaging.cc
+76-4Lines changed: 76 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,14 +41,27 @@ namespace {
4141
// `MessagePort`s and `SharedArrayBuffer`s, and make new JS objects out of them.
4242
class DeserializerDelegate : public ValueDeserializer::Delegate {
4343
public:
44-
DeserializerDelegate(Message* m, Environment* env)
45-
: env_(env), msg_(m) {}
44+
DeserializerDelegate(Message* m,
45+
Environment* env,
46+
const std::vector<MessagePort*>& message_ports)
47+
: env_(env), msg_(m), message_ports_(message_ports) {}
48+
49+
MaybeLocal<Object> ReadHostObject(Isolate* isolate) override {
50+
// Currently, only MessagePort hosts objects are supported, so identifying
51+
// by the index in the message's MessagePort array is sufficient.
52+
uint32_t id;
53+
if (!deserializer->ReadUint32(&id))
54+
return MaybeLocal<Object>();
55+
CHECK_LE(id, message_ports_.size());
56+
return message_ports_[id]->object();
57+
};
4658

4759
ValueDeserializer* deserializer = nullptr;
4860

4961
private:
5062
Environment* env_;
5163
Message* msg_;
64+
const std::vector<MessagePort*>& message_ports_;
5265
};
5366

5467
} // anonymous namespace
@@ -58,7 +71,23 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
5871
EscapableHandleScope handle_scope(env->isolate());
5972
Context::Scope context_scope(context);
6073

61-
DeserializerDelegate delegate(this, env);
74+
// Create all necessary MessagePort handles.
75+
std::vector<MessagePort*> ports(message_ports_.size());
76+
for (uint32_t i = 0; i < message_ports_.size(); ++i) {
77+
ports[i] = MessagePort::New(env,
78+
context,
79+
std::move(message_ports_[i]));
80+
if (ports[i] == nullptr) {
81+
for (MessagePort* port : ports) {
82+
// This will eventually release the MessagePort object itself.
83+
port->Close();
84+
}
85+
return MaybeLocal<Value>();
86+
}
87+
}
88+
message_ports_.clear();
89+
90+
DeserializerDelegate delegate(this, env, ports);
6291
ValueDeserializer deserializer(
6392
env->isolate(),
6493
reinterpret_cast<const uint8_t*>(main_message_buf_.data),
@@ -83,6 +112,10 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
83112
deserializer.ReadValue(context).FromMaybe(Local<Value>()));
84113
}
85114

115+
void Message::AddMessagePort(std::unique_ptr<MessagePortData>&& data) {
116+
message_ports_.emplace_back(std::move(data));
117+
}
118+
86119
namespace {
87120

88121
// This tells V8 how to serialize objects that it does not understand
@@ -97,12 +130,43 @@ class SerializerDelegate : public ValueSerializer::Delegate {
97130
env_->isolate()->ThrowException(Exception::Error(message));
98131
}
99132

133+
Maybe<bool> WriteHostObject(Isolate* isolate, Local<Object> object) override {
134+
if (env_->message_port_constructor_template()->HasInstance(object)) {
135+
return WriteMessagePort(Unwrap<MessagePort>(object));
136+
}
137+
138+
THROW_ERR_CANNOT_TRANSFER_OBJECT(env_);
139+
return Nothing<bool>();
140+
}
141+
142+
void Finish() {
143+
// Only close the MessagePort handles and actually transfer them
144+
// once we know that serialization succeeded.
145+
for (MessagePort* port : ports_) {
146+
port->Close();
147+
msg_->AddMessagePort(port->Detach());
148+
}
149+
}
150+
100151
ValueSerializer* serializer = nullptr;
101152

102153
private:
154+
Maybe<bool> WriteMessagePort(MessagePort* port) {
155+
for (uint32_t i = 0; i < ports_.size(); i++) {
156+
if (ports_[i] == port) {
157+
serializer->WriteUint32(i);
158+
return Just(true);
159+
}
160+
}
161+
162+
THROW_ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST(env_);
163+
return Nothing<bool>();
164+
}
165+
103166
Environment* env_;
104167
Local<Context> context_;
105168
Message* msg_;
169+
std::vector<MessagePort*> ports_;
106170

107171
friend class worker::Message;
108172
};
@@ -131,7 +195,7 @@ Maybe<bool> Message::Serialize(Environment* env,
131195
Local<Value> entry;
132196
if (!transfer_list->Get(context, i).ToLocal(&entry))
133197
return Nothing<bool>();
134-
// Currently, we support ArrayBuffers.
198+
// Currently, we support ArrayBuffers and MessagePorts.
135199
if (entry->IsArrayBuffer()) {
136200
Local<ArrayBuffer> ab = entry.As<ArrayBuffer>();
137201
// If we cannot render the ArrayBuffer unusable in this Isolate and
@@ -144,6 +208,12 @@ Maybe<bool> Message::Serialize(Environment* env,
144208
array_buffers.push_back(ab);
145209
serializer.TransferArrayBuffer(id, ab);
146210
continue;
211+
} else if (env->message_port_constructor_template()
212+
->HasInstance(entry)) {
213+
MessagePort* port = Unwrap<MessagePort>(entry.As<Object>());
214+
CHECK_NE(port, nullptr);
215+
delegate.ports_.push_back(port);
216+
continue;
147217
}
148218

149219
THROW_ERR_INVALID_TRANSFER_OBJECT(env);
@@ -167,6 +237,8 @@ Maybe<bool> Message::Serialize(Environment* env,
167237
contents.ByteLength() });
168238
}
169239

240+
delegate.Finish();
241+
170242
// The serializer gave us a buffer allocated using `malloc()`.
171243
std::pair<uint8_t*, size_t> data = serializer.Release();
172244
main_message_buf_ =
Collapse file

‎src/node_messaging.h‎

Copy file name to clipboardExpand all lines: src/node_messaging.h
+5Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,9 +37,14 @@ class Message {
3737
v8::Local<v8::Value> input,
3838
v8::Local<v8::Value> transfer_list);
3939

40+
// Internal method of Message that is called once serialization finishes
41+
// and that transfers ownership of `data` to this message.
42+
void AddMessagePort(std::unique_ptr<MessagePortData>&& data);
43+
4044
private:
4145
MallocedBuffer<char> main_message_buf_;
4246
std::vector<MallocedBuffer<char>> array_buffer_contents_;
47+
std::vector<std::unique_ptr<MessagePortData>> message_ports_;
4348

4449
friend class MessagePort;
4550
};
Collapse file

‎test/parallel/test-message-port-message-port-transferring.js‎

Copy file name to clipboard
+23Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
// Flags: --experimental-worker
2+
'use strict';
3+
const common = require('../common');
4+
const assert = require('assert');
5+
6+
const { MessageChannel } = require('worker');
7+
8+
{
9+
const { port1: basePort1, port2: basePort2 } = new MessageChannel();
10+
const {
11+
port1: transferredPort1, port2: transferredPort2
12+
} = new MessageChannel();
13+
14+
basePort1.postMessage({ transferredPort1 }, [ transferredPort1 ]);
15+
basePort2.on('message', common.mustCall(({ transferredPort1 }) => {
16+
transferredPort1.postMessage('foobar');
17+
transferredPort2.on('message', common.mustCall((msg) => {
18+
assert.strictEqual(msg, 'foobar');
19+
transferredPort1.close(common.mustCall());
20+
basePort1.close(common.mustCall());
21+
}));
22+
}));
23+
}

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.