vector_core/
rt.rs

1//! Task and timer facade: tokio natively, the browser event loop on wasm32.
2//!
3//! wasm32-unknown-unknown has no threads and no tokio driver, so spawns go to
4//! `spawn_local` and timers to `setTimeout`. Signatures match tokio's so call
5//! sites read the same on both targets.
6
7#[cfg(not(target_arch = "wasm32"))]
8pub use tokio::task::{spawn, spawn_blocking, yield_now, JoinError, JoinHandle};
9
10#[cfg(not(target_arch = "wasm32"))]
11pub mod time {
12    pub use tokio::time::*;
13}
14
15#[cfg(target_arch = "wasm32")]
16pub use wasm::*;
17
18/// Whether work can be spawned from here: natively only inside a tokio
19/// runtime (a drop off-runtime, or during shutdown, has none); always in the
20/// browser, whose event loop is the runtime.
21pub fn can_spawn() -> bool {
22    #[cfg(not(target_arch = "wasm32"))]
23    {
24        tokio::runtime::Handle::try_current().is_ok()
25    }
26    #[cfg(target_arch = "wasm32")]
27    {
28        true
29    }
30}
31
32#[cfg(target_arch = "wasm32")]
33mod wasm {
34    use futures_util::future::{AbortHandle, Abortable};
35    use std::future::Future;
36    use std::pin::Pin;
37    use std::sync::atomic::{AtomicBool, Ordering};
38    use std::sync::Arc;
39    use std::task::{Context, Poll};
40
41    pub mod time {
42        pub use std::time::Duration;
43        pub use wasmtimer::std::Instant;
44        pub use wasmtimer::tokio::{
45            interval, interval_at, sleep, sleep_until, Interval, MissedTickBehavior, Sleep, Timeout,
46        };
47
48        // tokio's take `IntoFuture`, which nostr-sdk's request builders rely on.
49        pub fn timeout<F: std::future::IntoFuture>(d: Duration, f: F) -> Timeout<F::IntoFuture> {
50            wasmtimer::tokio::timeout(d, f.into_future())
51        }
52
53        pub fn timeout_at<F: std::future::IntoFuture>(at: Instant, f: F) -> Timeout<F::IntoFuture> {
54            wasmtimer::tokio::timeout_at(at, f.into_future())
55        }
56        pub mod error {
57            pub use wasmtimer::tokio::error::Elapsed;
58        }
59    }
60
61    #[derive(Debug)]
62    pub struct JoinError;
63
64    impl JoinError {
65        pub fn is_cancelled(&self) -> bool { true }
66        pub fn is_panic(&self) -> bool { false }
67    }
68
69    impl std::fmt::Display for JoinError {
70        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71            f.write_str("task was cancelled")
72        }
73    }
74
75    impl std::error::Error for JoinError {}
76
77    pub struct JoinHandle<T> {
78        rx: tokio::sync::oneshot::Receiver<T>,
79        abort: AbortHandle,
80        done: Arc<AtomicBool>,
81    }
82
83    impl<T> JoinHandle<T> {
84        pub fn abort(&self) { self.abort.abort(); }
85        pub fn is_finished(&self) -> bool { self.done.load(Ordering::Relaxed) || self.abort.is_aborted() }
86    }
87
88    impl<T> Future for JoinHandle<T> {
89        type Output = Result<T, JoinError>;
90        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
91            Pin::new(&mut self.rx).poll(cx).map(|r| r.map_err(|_| JoinError))
92        }
93    }
94
95    pub fn spawn<F>(fut: F) -> JoinHandle<F::Output>
96    where
97        F: Future + 'static,
98        F::Output: 'static,
99    {
100        let (tx, rx) = tokio::sync::oneshot::channel();
101        let (abort, reg) = AbortHandle::new_pair();
102        let done = Arc::new(AtomicBool::new(false));
103        let flag = done.clone();
104        wasm_bindgen_futures::spawn_local(async move {
105            if let Ok(v) = Abortable::new(fut, reg).await {
106                let _ = tx.send(v);
107            }
108            flag.store(true, Ordering::Relaxed);
109        });
110        JoinHandle { rx, abort, done }
111    }
112
113    /// No worker pool: the closure runs on the next tick of the event loop.
114    pub fn spawn_blocking<F, R>(f: F) -> JoinHandle<R>
115    where
116        F: FnOnce() -> R + 'static,
117        R: 'static,
118    {
119        spawn(async move { f() })
120    }
121
122    pub async fn yield_now() {
123        struct YieldNow(bool);
124        impl Future for YieldNow {
125            type Output = ();
126            fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
127                if self.0 {
128                    return Poll::Ready(());
129                }
130                self.0 = true;
131                cx.waker().wake_by_ref();
132                Poll::Pending
133            }
134        }
135        YieldNow(false).await
136    }
137}
138
139/// `Send` natively; nothing on wasm32, where there is one thread and browser
140/// futures hold `!Send` JS handles.
141#[cfg(not(target_arch = "wasm32"))]
142pub trait MaybeSend: Send {}
143#[cfg(not(target_arch = "wasm32"))]
144impl<T: Send + ?Sized> MaybeSend for T {}
145
146#[cfg(target_arch = "wasm32")]
147pub trait MaybeSend {}
148#[cfg(target_arch = "wasm32")]
149impl<T: ?Sized> MaybeSend for T {}
150
151/// Wall clock and `Instant` that work on both targets, for `#[macro_export]`
152/// macros that expand in crates without a `web-time` dependency.
153pub use web_time as time_std;
vector_core/rt.rs

vector_core/
rt.rs
