anyswap_sdk/driver/
rt.rs

1//! The runtime's shape: the calls the driver makes on it and the thread bounds
2//! it asks for. Tokio and `Send` natively, the browser's own and no bounds on
3//! wasm32, so the loop is written once. Nothing else in the driver may name
4//! a runtime.
5
6use std::{future::Future, time::Duration};
7
8/// `Send` everywhere but wasm32, where nothing crosses a thread and the
9/// browser's objects cannot.
10#[cfg(not(target_arch = "wasm32"))]
11pub trait MaybeSend: Send {}
12#[cfg(not(target_arch = "wasm32"))]
13impl<T: Send + ?Sized> MaybeSend for T {}
14#[cfg(target_arch = "wasm32")]
15pub trait MaybeSend {}
16#[cfg(target_arch = "wasm32")]
17impl<T: ?Sized> MaybeSend for T {}
18
19/// `Sync` on the same terms as [`MaybeSend`].
20#[cfg(not(target_arch = "wasm32"))]
21pub trait MaybeSync: Sync {}
22#[cfg(not(target_arch = "wasm32"))]
23impl<T: Sync + ?Sized> MaybeSync for T {}
24#[cfg(target_arch = "wasm32")]
25pub trait MaybeSync {}
26#[cfg(target_arch = "wasm32")]
27impl<T: ?Sized> MaybeSync for T {}
28
29/// Spawns a future on the runtime the driver runs on: tokio natively, the
30/// browser's event loop on wasm32. What `Driver::run` is handed to, written
31/// once for both targets.
32pub fn spawn(future: impl Future<Output = ()> + MaybeSend + 'static) {
33    #[cfg(not(target_arch = "wasm32"))]
34    tokio::spawn(future);
35    #[cfg(target_arch = "wasm32")]
36    wasm_bindgen_futures::spawn_local(future);
37}
38
39#[cfg(not(target_arch = "wasm32"))]
40pub(crate) async fn sleep(duration: Duration) {
41    tokio::time::sleep(duration).await;
42}
43
44#[cfg(target_arch = "wasm32")]
45#[wasm_bindgen::prelude::wasm_bindgen]
46extern "C" {
47    /// The global `setTimeout`, so this works in a worker as well as a window.
48    #[wasm_bindgen(js_name = setTimeout)]
49    fn set_timeout(handler: &js_sys::Function, millis: i32);
50}
51
52#[cfg(target_arch = "wasm32")]
53pub(crate) async fn sleep(duration: Duration) {
54    let millis = duration.as_millis().min(i32::MAX as u128) as i32;
55    let promise = js_sys::Promise::new(&mut |resolve, _| set_timeout(&resolve, millis));
56    let _ = wasm_bindgen_futures::JsFuture::from(promise).await;
57}
58
59/// The error a request ends in when it goes unanswered for `limit`.
60pub(crate) trait Timeout {
61    fn timeout(limit: Duration) -> Self;
62}
63
64/// Held while one swap's record is read, passed, and written, against anything
65/// else sharing its store. On wasm32 it is a Web Lock named `name`, so tabs
66/// over one IndexedDB take turns; natively it is nothing, since one process
67/// owns a store.
68#[cfg(not(target_arch = "wasm32"))]
69pub(crate) struct Exclusive;
70
71#[cfg(not(target_arch = "wasm32"))]
72pub(crate) async fn exclusive(_name: &str) -> Exclusive {
73    Exclusive
74}
75
76#[cfg(target_arch = "wasm32")]
77pub(crate) struct Exclusive(Option<js_sys::Function>);
78
79#[cfg(target_arch = "wasm32")]
80impl Drop for Exclusive {
81    fn drop(&mut self) {
82        if let Some(release) = self.0.take() {
83            let _ = release.call0(&wasm_bindgen::JsValue::UNDEFINED);
84        }
85    }
86}
87
88/// Holds nothing where `navigator.locks` is missing, as in Node or outside a
89/// secure context, or refuses the request. Dropping the guard settles the
90/// promise the lock is held on, so a lock granted after its waiter is gone
91/// is let go at once.
92#[cfg(target_arch = "wasm32")]
93pub(crate) async fn exclusive(name: &str) -> Exclusive {
94    use js_sys::{Array, Function, Promise, Reflect};
95    use wasm_bindgen::{JsCast, JsValue, closure::Closure};
96
97    let locks = Reflect::get(&js_sys::global(), &"navigator".into())
98        .and_then(|navigator| Reflect::get(&navigator, &"locks".into()))
99        .ok()
100        .filter(JsValue::is_object);
101    let Some(locks) = locks else {
102        return Exclusive(None);
103    };
104    let Some(request) = Reflect::get(&locks, &"request".into())
105        .ok()
106        .and_then(|request| request.dyn_into::<Function>().ok())
107    else {
108        return Exclusive(None);
109    };
110
111    let mut release = None;
112    let held = Promise::new(&mut |resolve, _| release = Some(resolve));
113    let guard = Exclusive(release);
114    let mut grant = None;
115    let granted = Promise::new(&mut |resolve, _| grant = Some(resolve));
116    let Some(grant) = grant else {
117        return guard;
118    };
119    let callback = Closure::once_into_js(move |_lock: JsValue| {
120        let _ = grant.call0(&JsValue::UNDEFINED);
121        held
122    });
123    let Ok(pending) = request.call2(&locks, &JsValue::from_str(name), &callback) else {
124        return guard;
125    };
126    // `pending` settles once the lock is let go, or at once if refused.
127    let pending = Promise::resolve(&pending);
128    let _ =
129        wasm_bindgen_futures::JsFuture::from(Promise::race(&Array::of2(&granted, &pending))).await;
130    guard
131}
132
133/// `call`'s result, or its timeout once `limit` passes first, so a peer that
134/// never answers costs a pass its wait and nothing more.
135pub(crate) async fn bounded<T, E: Timeout>(
136    limit: Duration,
137    call: impl Future<Output = Result<T, E>>,
138) -> Result<T, E> {
139    tokio::select! {
140        result = call => result,
141        _ = sleep(limit) => Err(E::timeout(limit)),
142    }
143}
anyswap_sdk/driver/rt.rs

anyswap_sdk/driver/
rt.rs
