1use n0_future::{StreamExt, task::AbortOnDropHandle};2use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode};3use napi_derive::napi;4use tokio::sync::Mutex;56use crate::{EndpointAddr, PathEvent, PathSnapshot};78/// Handle to a running watcher task. Call `stop()` (or drop) to unregister.910pub struct WatchHandle {11 task: Mutex<Option<AbortOnDropHandle<()>>>,12}1314impl WatchHandle {15 pub(crate) fn new(task: AbortOnDropHandle<()>) -> Self {16 Self {17 task: Mutex::new(Some(task)),18 }19 }20}212223impl WatchHandle {24 /// Stop the watcher, aborting the background task.25 26 pub async fn stop(&self) {27 self.task.lock().await.take();28 }29}3031pub(crate) fn spawn_watch_addr(32 endpoint: iroh::Endpoint,33 cb: ThreadsafeFunction<EndpointAddr>,34) -> WatchHandle {35 let task = n0_future::task::spawn(async move {36 use iroh::Watcher;37 let mut stream = endpoint.watch_addr().stream();38 while let Some(addr) = stream.next().await {39 let mapped: EndpointAddr = addr.into();40 cb.call(Ok(mapped), ThreadsafeFunctionCallMode::NonBlocking);41 }42 });43 WatchHandle::new(AbortOnDropHandle::new(task))44}4546pub(crate) fn spawn_home_relay_watch(47 endpoint: iroh::Endpoint,48 cb: ThreadsafeFunction<Vec<String>>,49) -> WatchHandle {50 let task = n0_future::task::spawn(async move {51 use iroh::Watcher;52 let mut stream = endpoint.home_relay_status().stream();53 while let Some(statuses) = stream.next().await {54 let urls: Vec<String> = statuses.into_iter().map(|s| s.url().to_string()).collect();55 cb.call(Ok(urls), ThreadsafeFunctionCallMode::NonBlocking);56 }57 });58 WatchHandle::new(AbortOnDropHandle::new(task))59}6061pub(crate) fn spawn_network_change_watch(62 endpoint: iroh::Endpoint,63 cb: ThreadsafeFunction<()>,64) -> WatchHandle {65 let task = n0_future::task::spawn(async move {66 loop {67 endpoint.network_change().await;68 cb.call(Ok(()), ThreadsafeFunctionCallMode::NonBlocking);69 }70 });71 WatchHandle::new(AbortOnDropHandle::new(task))72}7374pub(crate) fn spawn_paths_watch(75 conn: iroh::endpoint::Connection,76 cb: ThreadsafeFunction<Vec<PathSnapshot>>,77) -> WatchHandle {78 let task = n0_future::task::spawn(async move {79 let mut stream = conn.paths_stream();80 while let Some(snapshot) = stream.next().await {81 let mapped: Vec<PathSnapshot> = snapshot82 .iter()83 .map(|p| PathSnapshot {84 id: p.id().to_string(),85 is_selected: p.is_selected(),86 remote_addr: crate::path::transport_addr_to_string(p.remote_addr()),87 is_ip: p.is_ip(),88 is_relay: p.is_relay(),89 rtt_ms: p.rtt().as_millis() as i64,90 stats: p.stats().into(),91 })92 .collect();93 cb.call(Ok(mapped), ThreadsafeFunctionCallMode::NonBlocking);94 }95 });96 WatchHandle::new(AbortOnDropHandle::new(task))97}9899pub(crate) fn spawn_path_events_watch(100 conn: iroh::endpoint::Connection,101 cb: ThreadsafeFunction<PathEvent>,102) -> WatchHandle {103 let task = n0_future::task::spawn(async move {104 let mut stream = conn.path_events();105 while let Some(event) = stream.next().await {106 let mapped: PathEvent = event.into();107 cb.call(Ok(mapped), ThreadsafeFunctionCallMode::NonBlocking);108 }109 });110 WatchHandle::new(AbortOnDropHandle::new(task))111}