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 refuses
89/// the request; browsers leave it out of a page that is not a secure context,
90/// which the wasm bindings refuse to start in. Dropping the guard settles the
91/// promise the lock is held on, so a lock granted after its waiter is gone
92/// is let go at once.
93#[cfg(target_arch = "wasm32")]
94pub(crate) async fn exclusive(name: &str) -> Exclusive {
95    use js_sys::{Array, Function, Promise, Reflect};
96    use wasm_bindgen::{JsCast, JsValue, closure::Closure};
97
98    let locks = Reflect::get(&js_sys::global(), &"navigator".into())
99        .and_then(|navigator| Reflect::get(&navigator, &"locks".into()))
100        .ok()
101        .filter(JsValue::is_object);
102    let Some(locks) = locks else {
103        return Exclusive(None);
104    };
105    let Some(request) = Reflect::get(&locks, &"request".into())
106        .ok()
107        .and_then(|request| request.dyn_into::<Function>().ok())
108    else {
109        return Exclusive(None);
110    };
111
112    let mut release = None;
113    let held = Promise::new(&mut |resolve, _| release = Some(resolve));
114    let guard = Exclusive(release);
115    let mut grant = None;
116    let granted = Promise::new(&mut |resolve, _| grant = Some(resolve));
117    let Some(grant) = grant else {
118        return guard;
119    };
120    let callback = Closure::once_into_js(move |_lock: JsValue| {
121        let _ = grant.call0(&JsValue::UNDEFINED);
122        held
123    });
124    let Ok(pending) = request.call2(&locks, &JsValue::from_str(name), &callback) else {
125        return guard;
126    };
127    // `pending` settles once the lock is let go, or at once if refused.
128    let pending = Promise::resolve(&pending);
129    let _ =
130        wasm_bindgen_futures::JsFuture::from(Promise::race(&Array::of2(&granted, &pending))).await;
131    guard
132}
133
134/// `call`'s result, or its timeout once `limit` passes first, so a peer that
135/// never answers costs a pass its wait and nothing more.
136pub(crate) async fn bounded<T, E: Timeout>(
137    limit: Duration,
138    call: impl Future<Output = Result<T, E>>,
139) -> Result<T, E> {
140    tokio::select! {
141        result = call => result,
142        _ = sleep(limit) => Err(E::timeout(limit)),
143    }
144}
anyswap_sdk/driver/rt.rs

anyswap_sdk/driver/
rt.rs
