diff --git a/changelog.d/8327-worker-channel-listeners.md b/changelog.d/8327-worker-channel-listeners.md new file mode 100644 index 0000000000..7436d01f52 --- /dev/null +++ b/changelog.d/8327-worker-channel-listeners.md @@ -0,0 +1,8 @@ +### Fixed + +- **Worker channel listeners now preserve Node registration semantics (#6763).** + `MessagePort` keeps distinct listeners in order, deduplicates repeated + registrations, removes only the requested callback, honors `once`, reports + listener counts, and passes the close event to close listeners. + `MessagePort` and `BroadcastChannel` EventTarget listeners also honor the + `{ once: true }` option, with callback snapshots rooted across moving GC. diff --git a/crates/perry-stdlib/src/worker_threads.rs b/crates/perry-stdlib/src/worker_threads.rs index d82efc80fa..47060e755c 100644 --- a/crates/perry-stdlib/src/worker_threads.rs +++ b/crates/perry-stdlib/src/worker_threads.rs @@ -124,14 +124,14 @@ struct MessagePortState { object_bits: u64, /// Queue of delivered structured-clone snapshots (oldest first). inbox: VecDeque, - /// `message` event listener (NaN-boxed closure value bits), if registered. - message_cb: Option, - /// `close` event listener (NaN-boxed closure value bits), if registered. - close_cb: Option, + /// Node-style `message` listeners registered through on()/once(). + message_cbs: Vec, + /// Node-style `close` listeners registered through on()/once(). + close_cbs: Vec, /// `message` listeners registered through addEventListener(). - message_event_cbs: Vec, + message_event_cbs: Vec, /// `close` listeners registered through addEventListener(). - close_event_cbs: Vec, + close_event_cbs: Vec, /// Whether `.start()` (or a `message` listener) has been attached. Until a /// port is started, queued messages are not dispatched to the listener /// (Node semantics), though `receiveMessageOnPort` still drains them. @@ -152,11 +152,17 @@ struct BroadcastChannelState { /// Queue of delivered structured-clone snapshots (oldest first). inbox: VecDeque, /// `message` listeners registered through addEventListener(). - message_event_cbs: Vec, + message_event_cbs: Vec, /// Whether `close()` has detached this BroadcastChannel. closed: bool, } +#[derive(Clone, Copy)] +struct EventListener { + callback_bits: u64, + once: bool, +} + static ENVIRONMENT_DATA_GC_REGISTERED: Once = Once::new(); static WORKER_GC_REGISTERED: Once = Once::new(); static NEXT_WORKER_ID: AtomicU64 = AtomicU64::new(1); @@ -230,26 +236,27 @@ fn scan_environment_data_roots_mut(visitor: &mut perry_runtime::gc::RuntimeRootV MESSAGE_PORTS.with(|ports| { for state in ports.borrow_mut().values_mut() { visitor.visit_nanbox_u64_slot(&mut state.object_bits); - if let Some(cb) = state.message_cb.as_mut() { - visitor.visit_nanbox_u64_slot(cb); - } - if let Some(cb) = state.close_cb.as_mut() { - visitor.visit_nanbox_u64_slot(cb); + for listener in state + .message_cbs + .iter_mut() + .chain(state.close_cbs.iter_mut()) + { + visitor.visit_nanbox_u64_slot(&mut listener.callback_bits); } - for cb in state + for listener in state .message_event_cbs .iter_mut() .chain(state.close_event_cbs.iter_mut()) { - visitor.visit_nanbox_u64_slot(cb); + visitor.visit_nanbox_u64_slot(&mut listener.callback_bits); } } }); BROADCAST_CHANNELS.with(|channels| { for state in channels.borrow_mut().values_mut() { visitor.visit_nanbox_u64_slot(&mut state.object_bits); - for cb in state.message_event_cbs.iter_mut() { - visitor.visit_nanbox_u64_slot(cb); + for listener in state.message_event_cbs.iter_mut() { + visitor.visit_nanbox_u64_slot(&mut listener.callback_bits); } } }); @@ -478,6 +485,13 @@ fn callback_bits_from_value(value: f64) -> Option { perry_runtime::closure::is_closure_ptr(ptr).then_some(bits) } +fn listener_once(options: f64) -> bool { + let Some(options) = object_ptr_from_value(options) else { + return false; + }; + perry_runtime::value::js_is_truthy(get_object_field(options, "once")) != 0 +} + extern "C" fn worker_threads_noop0(_closure: *const ClosureHeader) -> f64 { js_undefined() } @@ -691,16 +705,6 @@ fn message_value_is_uncloneable(value: f64, visited: &mut HashSet) -> boo }) } -fn call_callback0(callback_bits: u64, this_bits: u64) { - let closure = closure_ptr_from_bits(callback_bits); - if closure.is_null() { - return; - } - let prev_this = perry_runtime::object::js_implicit_this_set(f64::from_bits(this_bits)); - perry_runtime::closure::js_closure_call0(closure); - perry_runtime::object::js_implicit_this_set(prev_this); -} - fn call_callback1(callback_bits: u64, this_bits: u64, arg: f64) { let closure = closure_ptr_from_bits(callback_bits); if closure.is_null() { diff --git a/crates/perry-stdlib/src/worker_threads/broadcast_channel.rs b/crates/perry-stdlib/src/worker_threads/broadcast_channel.rs index d121cf81a7..a6cd495871 100644 --- a/crates/perry-stdlib/src/worker_threads/broadcast_channel.rs +++ b/crates/perry-stdlib/src/worker_threads/broadcast_channel.rs @@ -58,6 +58,7 @@ extern "C" fn broadcast_add_event_listener( closure: *const ClosureHeader, event: f64, callback: f64, + options: f64, ) -> f64 { let channel_id = port_id_from_closure(closure); let event_name = string_value_to_string(event).unwrap_or_default(); @@ -67,8 +68,16 @@ extern "C" fn broadcast_add_event_listener( super::async_shim::ensure_pump_registered(); BROADCAST_CHANNELS.with(|channels| { if let Some(state) = channels.borrow_mut().get_mut(&channel_id) { - if event_name == "message" && !state.message_event_cbs.contains(&cb_bits) { - state.message_event_cbs.push(cb_bits); + if event_name == "message" + && !state + .message_event_cbs + .iter() + .any(|listener| listener.callback_bits == cb_bits) + { + state.message_event_cbs.push(EventListener { + callback_bits: cb_bits, + once: listener_once(options), + }); } } }); @@ -88,7 +97,9 @@ extern "C" fn broadcast_remove_event_listener( BROADCAST_CHANNELS.with(|channels| { if let Some(state) = channels.borrow_mut().get_mut(&channel_id) { if event_name == "message" { - state.message_event_cbs.retain(|cb| *cb != cb_bits); + state + .message_event_cbs + .retain(|listener| listener.callback_bits != cb_bits); } } }); @@ -139,7 +150,7 @@ pub extern "C" fn js_worker_threads_broadcast_channel_new(name: f64) -> f64 { set_object_field( obj, "addEventListener", - port_bound_closure(broadcast_add_event_listener as *const u8, 2, id), + port_bound_closure(broadcast_add_event_listener as *const u8, 3, id), ); set_object_field( obj, diff --git a/crates/perry-stdlib/src/worker_threads/channel_pump.rs b/crates/perry-stdlib/src/worker_threads/channel_pump.rs index af1dc0c122..9f0dcc49a0 100644 --- a/crates/perry-stdlib/src/worker_threads/channel_pump.rs +++ b/crates/perry-stdlib/src/worker_threads/channel_pump.rs @@ -6,7 +6,7 @@ //! `crate::worker_threads::js_worker_threads_channels_*` keeps resolving. use super::{ - call_callback0, call_callback1, deserialize_message, event_object, object_event_handler, + call_callback1, deserialize_message, event_object, object_event_handler, EventListener, SerializedMessage, BROADCAST_CHANNELS, MESSAGE_PORTS, }; @@ -22,8 +22,8 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { // postMessage / close, which needs to borrow MESSAGE_PORTS again. struct MessageDispatch { target_bits: u64, - raw_cb: Option, - event_cbs: Vec, + raw_cbs: Vec, + event_cbs: Vec, handler_cb: Option, msg: SerializedMessage, } @@ -46,16 +46,22 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { let mut ports = ports.borrow_mut(); let state = ports.get_mut(&port_id)?; let has_event_target = state.started - && (state.message_cb.is_some() || !state.message_event_cbs.is_empty()); + && (!state.message_cbs.is_empty() || !state.message_event_cbs.is_empty()); if state.closed || (!has_event_target && handler_cb.is_none()) { return None; } - state.inbox.pop_front().map(|msg| MessageDispatch { - target_bits: state.object_bits, - raw_cb: state.message_cb, - event_cbs: state.message_event_cbs.clone(), - handler_cb, - msg, + state.inbox.pop_front().map(|msg| { + let raw_cbs = state.message_cbs.clone(); + state.message_cbs.retain(|listener| !listener.once); + let event_cbs = state.message_event_cbs.clone(); + state.message_event_cbs.retain(|listener| !listener.once); + MessageDispatch { + target_bits: state.object_bits, + raw_cbs, + event_cbs, + handler_cb, + msg, + } }) }); if next.is_some() { @@ -64,17 +70,50 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { } match next { Some(dispatch) => { + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let target = scope.root_nanbox_f64(f64::from_bits(dispatch.target_bits)); + let raw_cbs = dispatch + .raw_cbs + .into_iter() + .map(|listener| scope.root_nanbox_f64(f64::from_bits(listener.callback_bits))) + .collect::>(); + let event_cbs = dispatch + .event_cbs + .into_iter() + .map(|listener| scope.root_nanbox_f64(f64::from_bits(listener.callback_bits))) + .collect::>(); + let handler_cb = dispatch + .handler_cb + .map(|bits| scope.root_nanbox_f64(f64::from_bits(bits))); let value = deserialize_message(&dispatch.msg); - if let Some(cb_bits) = dispatch.raw_cb { - call_callback1(cb_bits, dispatch.target_bits, value); + let value = scope.root_nanbox_f64(value); + for callback in raw_cbs { + call_callback1( + callback.get_nanbox_f64().to_bits(), + target.get_nanbox_f64().to_bits(), + value.get_nanbox_f64(), + ); } - if !dispatch.event_cbs.is_empty() || dispatch.handler_cb.is_some() { - let event = event_object("message", dispatch.target_bits, Some(value)); - for cb_bits in dispatch.event_cbs { - call_callback1(cb_bits, dispatch.target_bits, event); + if !event_cbs.is_empty() || handler_cb.is_some() { + let event = event_object( + "message", + target.get_nanbox_f64().to_bits(), + Some(value.get_nanbox_f64()), + ); + let event = scope.root_nanbox_f64(event); + for callback in event_cbs { + call_callback1( + callback.get_nanbox_f64().to_bits(), + target.get_nanbox_f64().to_bits(), + event.get_nanbox_f64(), + ); } - if let Some(cb_bits) = dispatch.handler_cb { - call_callback1(cb_bits, dispatch.target_bits, event); + if let Some(callback) = handler_cb { + call_callback1( + callback.get_nanbox_f64().to_bits(), + target.get_nanbox_f64().to_bits(), + event.get_nanbox_f64(), + ); } } dispatched += 1; @@ -85,7 +124,7 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { struct BroadcastDispatch { target_bits: u64, - event_cbs: Vec, + event_cbs: Vec, handler_cb: Option, msg: SerializedMessage, } @@ -110,11 +149,15 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { if state.closed || (state.message_event_cbs.is_empty() && handler_cb.is_none()) { return None; } - state.inbox.pop_front().map(|msg| BroadcastDispatch { - target_bits: state.object_bits, - event_cbs: state.message_event_cbs.clone(), - handler_cb, - msg, + state.inbox.pop_front().map(|msg| { + let event_cbs = state.message_event_cbs.clone(); + state.message_event_cbs.retain(|listener| !listener.once); + BroadcastDispatch { + target_bits: state.object_bits, + event_cbs, + handler_cb, + msg, + } }) }); if next.is_some() { @@ -123,13 +166,37 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { } match next { Some(dispatch) => { + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let target = scope.root_nanbox_f64(f64::from_bits(dispatch.target_bits)); + let event_cbs = dispatch + .event_cbs + .into_iter() + .map(|listener| scope.root_nanbox_f64(f64::from_bits(listener.callback_bits))) + .collect::>(); + let handler_cb = dispatch + .handler_cb + .map(|bits| scope.root_nanbox_f64(f64::from_bits(bits))); let value = deserialize_message(&dispatch.msg); - let event = event_object("message", dispatch.target_bits, Some(value)); - if let Some(cb_bits) = dispatch.handler_cb { - call_callback1(cb_bits, dispatch.target_bits, event); + let value = scope.root_nanbox_f64(value); + let event = event_object( + "message", + target.get_nanbox_f64().to_bits(), + Some(value.get_nanbox_f64()), + ); + let event = scope.root_nanbox_f64(event); + if let Some(callback) = handler_cb { + call_callback1( + callback.get_nanbox_f64().to_bits(), + target.get_nanbox_f64().to_bits(), + event.get_nanbox_f64(), + ); } - for cb_bits in dispatch.event_cbs { - call_callback1(cb_bits, dispatch.target_bits, event); + for callback in event_cbs { + call_callback1( + callback.get_nanbox_f64().to_bits(), + target.get_nanbox_f64().to_bits(), + event.get_nanbox_f64(), + ); } dispatched += 1; } @@ -140,8 +207,8 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { // Fire `close` callbacks once for newly-closed ports. struct CloseDispatch { target_bits: u64, - raw_cb: Option, - event_cbs: Vec, + raw_cbs: Vec, + event_cbs: Vec, } let close_events: Vec = MESSAGE_PORTS.with(|ports| { @@ -149,23 +216,48 @@ pub extern "C" fn js_worker_threads_channels_process_pending() -> i32 { for state in ports.borrow_mut().values_mut() { if state.close_pending { state.close_pending = false; + let raw_cbs = state.close_cbs.clone(); + state.close_cbs.retain(|listener| !listener.once); + let event_cbs = state.close_event_cbs.clone(); + state.close_event_cbs.retain(|listener| !listener.once); events.push(CloseDispatch { target_bits: state.object_bits, - raw_cb: state.close_cb, - event_cbs: state.close_event_cbs.clone(), + raw_cbs, + event_cbs, }); } } events }); for event in close_events { - if let Some(cb_bits) = event.raw_cb { - call_callback0(cb_bits, event.target_bits); - } - if !event.event_cbs.is_empty() { - let close_event = event_object("close", event.target_bits, None); - for cb_bits in event.event_cbs { - call_callback1(cb_bits, event.target_bits, close_event); + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let target = scope.root_nanbox_f64(f64::from_bits(event.target_bits)); + let raw_cbs = event + .raw_cbs + .into_iter() + .map(|listener| scope.root_nanbox_f64(f64::from_bits(listener.callback_bits))) + .collect::>(); + let event_cbs = event + .event_cbs + .into_iter() + .map(|listener| scope.root_nanbox_f64(f64::from_bits(listener.callback_bits))) + .collect::>(); + if !raw_cbs.is_empty() || !event_cbs.is_empty() { + let close_event = event_object("close", target.get_nanbox_f64().to_bits(), None); + let close_event = scope.root_nanbox_f64(close_event); + for callback in raw_cbs { + call_callback1( + callback.get_nanbox_f64().to_bits(), + target.get_nanbox_f64().to_bits(), + close_event.get_nanbox_f64(), + ); + } + for callback in event_cbs { + call_callback1( + callback.get_nanbox_f64().to_bits(), + target.get_nanbox_f64().to_bits(), + close_event.get_nanbox_f64(), + ); } } dispatched += 1; @@ -181,7 +273,7 @@ pub extern "C" fn js_worker_threads_channels_has_pending() -> i32 { let pending_without_onmessage = MESSAGE_PORTS.with(|ports| { ports.borrow().values().any(|state| { let has_event_target = state.started - && (state.message_cb.is_some() || !state.message_event_cbs.is_empty()); + && (!state.message_cbs.is_empty() || !state.message_event_cbs.is_empty()); (!state.closed && !state.inbox.is_empty() && has_event_target) || state.close_pending }) }); diff --git a/crates/perry-stdlib/src/worker_threads/message_port.rs b/crates/perry-stdlib/src/worker_threads/message_port.rs index 5fa2ded0a7..6816bc91e1 100644 --- a/crates/perry-stdlib/src/worker_threads/message_port.rs +++ b/crates/perry-stdlib/src/worker_threads/message_port.rs @@ -33,7 +33,7 @@ pub(super) fn message_port_object(port_id: u64) -> *mut perry_runtime::object::O set_object_field( obj, "once", - port_bound_closure(port_on as *const u8, 2, port_id), + port_bound_closure(port_once as *const u8, 2, port_id), ); set_object_field( obj, @@ -45,10 +45,15 @@ pub(super) fn message_port_object(port_id: u64) -> *mut perry_runtime::object::O "removeListener", port_bound_closure(port_off as *const u8, 2, port_id), ); + set_object_field( + obj, + "listenerCount", + port_bound_closure(port_listener_count as *const u8, 1, port_id), + ); set_object_field( obj, "addEventListener", - port_bound_closure(port_add_event_listener as *const u8, 2, port_id), + port_bound_closure(port_add_event_listener as *const u8, 3, port_id), ); set_object_field( obj, @@ -121,15 +126,31 @@ extern "C" fn port_post_message(closure: *const ClosureHeader, value: f64, _tran js_undefined() } -/// port.on(event, callback) / addListener / once (#3157). +/// port.on(event, callback) / addListener (#3157). extern "C" fn port_on(closure: *const ClosureHeader, event: f64, callback: f64) -> f64 { + port_add_node_listener(closure, event, callback, false) +} + +/// port.once(event, callback) (#6763). +extern "C" fn port_once(closure: *const ClosureHeader, event: f64, callback: f64) -> f64 { + port_add_node_listener(closure, event, callback, true) +} + +fn port_add_node_listener( + closure: *const ClosureHeader, + event: f64, + callback: f64, + once: bool, +) -> f64 { let port_id = port_id_from_closure(closure); let event_name = string_value_to_string(event).unwrap_or_default(); if port_id == PARENT_PORT_HANDLE as u64 && CURRENT_WORKER_ID.with(|id| id.get()) != 0 { let callback_ptr = perry_runtime::value::js_nanbox_get_pointer(callback) as i64; return js_worker_threads_on(event.to_bits() as i64, callback_ptr); } - let cb_bits = callback.to_bits(); + let Some(cb_bits) = callback_bits_from_value(callback) else { + return js_undefined(); + }; // A program that only uses MessageChannel never calls spawn_for_promise, so // the runtime pump would otherwise never be registered and `main` would // return before any queued `message` is delivered. Register it here (mirrors @@ -139,11 +160,31 @@ extern "C" fn port_on(closure: *const ClosureHeader, event: f64, callback: f64) if let Some(state) = ports.borrow_mut().get_mut(&port_id) { match event_name.as_str() { "message" => { - state.message_cb = Some(cb_bits); + if !state + .message_cbs + .iter() + .any(|listener| listener.callback_bits == cb_bits) + { + state.message_cbs.push(EventListener { + callback_bits: cb_bits, + once, + }); + } // Attaching a `message` listener implicitly starts the port. state.started = true; } - "close" => state.close_cb = Some(cb_bits), + "close" => { + if !state + .close_cbs + .iter() + .any(|listener| listener.callback_bits == cb_bits) + { + state.close_cbs.push(EventListener { + callback_bits: cb_bits, + once, + }); + } + } _ => {} } } @@ -152,7 +193,7 @@ extern "C" fn port_on(closure: *const ClosureHeader, event: f64, callback: f64) } /// port.off(event) / removeListener (#3157). -extern "C" fn port_off(closure: *const ClosureHeader, event: f64, _callback: f64) -> f64 { +extern "C" fn port_off(closure: *const ClosureHeader, event: f64, callback: f64) -> f64 { let port_id = port_id_from_closure(closure); let event_name = string_value_to_string(event).unwrap_or_default(); if port_id == PARENT_PORT_HANDLE as u64 && CURRENT_WORKER_ID.with(|id| id.get()) != 0 { @@ -163,11 +204,18 @@ extern "C" fn port_off(closure: *const ClosureHeader, event: f64, _callback: f64 } return js_undefined(); } + let Some(cb_bits) = callback_bits_from_value(callback) else { + return js_undefined(); + }; MESSAGE_PORTS.with(|ports| { if let Some(state) = ports.borrow_mut().get_mut(&port_id) { match event_name.as_str() { - "message" => state.message_cb = None, - "close" => state.close_cb = None, + "message" => state + .message_cbs + .retain(|listener| listener.callback_bits != cb_bits), + "close" => state + .close_cbs + .retain(|listener| listener.callback_bits != cb_bits), _ => {} } } @@ -175,11 +223,28 @@ extern "C" fn port_off(closure: *const ClosureHeader, event: f64, _callback: f64 js_undefined() } +extern "C" fn port_listener_count(closure: *const ClosureHeader, event: f64) -> f64 { + let port_id = port_id_from_closure(closure); + let event_name = string_value_to_string(event).unwrap_or_default(); + MESSAGE_PORTS.with(|ports| { + let ports = ports.borrow(); + let Some(state) = ports.get(&port_id) else { + return 0.0; + }; + match event_name.as_str() { + "message" => (state.message_cbs.len() + state.message_event_cbs.len()) as f64, + "close" => (state.close_cbs.len() + state.close_event_cbs.len()) as f64, + _ => 0.0, + } + }) +} + /// port.addEventListener(event, callback) (#3598). extern "C" fn port_add_event_listener( closure: *const ClosureHeader, event: f64, callback: f64, + options: f64, ) -> f64 { let port_id = port_id_from_closure(closure); let event_name = string_value_to_string(event).unwrap_or_default(); @@ -192,13 +257,27 @@ extern "C" fn port_add_event_listener( match event_name.as_str() { "message" => { state.started = true; - if !state.message_event_cbs.contains(&cb_bits) { - state.message_event_cbs.push(cb_bits); + if !state + .message_event_cbs + .iter() + .any(|listener| listener.callback_bits == cb_bits) + { + state.message_event_cbs.push(EventListener { + callback_bits: cb_bits, + once: listener_once(options), + }); } } "close" => { - if !state.close_event_cbs.contains(&cb_bits) { - state.close_event_cbs.push(cb_bits); + if !state + .close_event_cbs + .iter() + .any(|listener| listener.callback_bits == cb_bits) + { + state.close_event_cbs.push(EventListener { + callback_bits: cb_bits, + once: listener_once(options), + }); } } _ => {} @@ -222,8 +301,12 @@ extern "C" fn port_remove_event_listener( MESSAGE_PORTS.with(|ports| { if let Some(state) = ports.borrow_mut().get_mut(&port_id) { match event_name.as_str() { - "message" => state.message_event_cbs.retain(|cb| *cb != cb_bits), - "close" => state.close_event_cbs.retain(|cb| *cb != cb_bits), + "message" => state + .message_event_cbs + .retain(|listener| listener.callback_bits != cb_bits), + "close" => state + .close_event_cbs + .retain(|listener| listener.callback_bits != cb_bits), _ => {} } } diff --git a/crates/perry/tests/issue_6763_broadcast_clone.rs b/crates/perry/tests/issue_6763_broadcast_clone.rs index 73656eddf9..371a0949f1 100644 --- a/crates/perry/tests/issue_6763_broadcast_clone.rs +++ b/crates/perry/tests/issue_6763_broadcast_clone.rs @@ -1,5 +1,5 @@ -//! Focused worker_threads structured-clone regressions from the #6763 parity -//! umbrella. These assertions keep the already-fixed increments in the normal +//! Focused worker_threads regressions from the #6763 parity umbrella. These +//! assertions keep the already-fixed increments in the normal //! integration suite instead of relying only on the full Node differential run. use std::path::PathBuf; @@ -168,3 +168,103 @@ channel.port2.close(); ) ); } + +#[test] +fn channel_event_target_once_listeners_are_removed_before_dispatch() { + let dir = tempfile::tempdir().expect("tempdir"); + let entry = dir.path().join("main.ts"); + let output = dir.path().join("main_bin"); + std::fs::write( + &entry, + r#" +import { BroadcastChannel, MessageChannel } from "node:worker_threads"; + +const channel = new MessageChannel(); +const nodePortEvents: string[] = []; +const portEvents: string[] = []; +function removed(value: string) { + nodePortEvents.push(`removed:${value}`); +} +channel.port2.on("message", removed); +channel.port2.off("message", removed); +channel.port2.once("message", (value) => nodePortEvents.push(`once:${value}`)); +channel.port2.on("message", (value) => { + nodePortEvents.push(`regular:${value}`); + if (value === "second") { + console.log("node port:", nodePortEvents.join(",")); + } +}); +channel.port2.on("close", (event) => console.log("close:", event.type)); +channel.port2.addEventListener( + "message", + (event) => portEvents.push(`once:${event.data}`), + { once: true }, +); +channel.port2.addEventListener("message", (event) => { + portEvents.push(`regular:${event.data}`); + if (event.data === "second") { + console.log("port:", portEvents.join(",")); + channel.port1.close(); + channel.port2.close(); + } +}); +channel.port1.postMessage("first"); +channel.port1.postMessage("second"); + +const sender = new BroadcastChannel("once-regression"); +const receiver = new BroadcastChannel("once-regression"); +const broadcastEvents: string[] = []; +receiver.addEventListener( + "message", + (event) => broadcastEvents.push(`once:${event.data}`), + { once: true }, +); +receiver.addEventListener("message", (event) => { + broadcastEvents.push(`regular:${event.data}`); + if (event.data === "second") { + console.log("broadcast:", broadcastEvents.join(",")); + sender.close(); + receiver.close(); + } +}); +sender.postMessage("first"); +sender.postMessage("second"); +"#, + ) + .expect("write fixture"); + + let compile = Command::new(perry_bin()) + .current_dir(dir.path()) + .arg("compile") + .arg(&entry) + .arg("-o") + .arg(&output) + .output() + .expect("run perry compile"); + assert!( + compile.status.success(), + "perry compile failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&compile.stdout), + String::from_utf8_lossy(&compile.stderr) + ); + + let run = Command::new(&output) + .output() + .expect("run compiled fixture"); + assert!( + run.status.success(), + "compiled fixture failed\nstatus: {:?}\nstdout:\n{}\nstderr:\n{}", + run.status, + String::from_utf8_lossy(&run.stdout), + String::from_utf8_lossy(&run.stderr) + ); + assert_eq!( + String::from_utf8_lossy(&run.stdout), + concat!( + "node port: once:first,regular:first,regular:second\n", + "port: once:first,regular:first,regular:second\n", + "broadcast: once:first,regular:first,regular:second\n", + "close: close\n", + ) + ); +}