cross_runtime/
task.rs

1#[cfg(not(target_arch = "wasm32"))]
2mod native {
3    use std::future::Future;
4
5    pub use tokio::task::{JoinError, JoinHandle};
6
7    /// Spawn a `Send` task on the current Tokio runtime.
8    pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
9    where
10        F: Future + Send + 'static,
11        F::Output: Send + 'static,
12    {
13        tokio::spawn(future)
14    }
15
16    /// Spawn a local task on the current Tokio `LocalSet`.
17    pub fn spawn_local<F>(future: F) -> JoinHandle<F::Output>
18    where
19        F: Future + 'static,
20        F::Output: 'static,
21    {
22        tokio::task::spawn_local(future)
23    }
24}
25
26#[cfg(target_arch = "wasm32")]
27mod web {
28    use futures_channel::oneshot;
29    use futures_util::future::{AbortHandle, Abortable};
30    use std::fmt;
31    use std::future::Future;
32    use std::pin::Pin;
33    use std::sync::Arc;
34    use std::sync::atomic::{AtomicBool, Ordering};
35    use std::task::{Context, Poll};
36
37    /// The task ended without producing its output.
38    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
39    pub struct JoinError;
40
41    impl JoinError {
42        /// Whether the task was cancelled.
43        #[must_use]
44        pub const fn is_cancelled(&self) -> bool {
45            true
46        }
47    }
48
49    impl fmt::Display for JoinError {
50        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51            f.write_str("task was cancelled")
52        }
53    }
54
55    impl std::error::Error for JoinError {}
56
57    /// A detached-on-drop task handle matching Tokio's lifecycle semantics.
58    pub struct JoinHandle<T> {
59        abort: AbortHandle,
60        result: oneshot::Receiver<Result<T, JoinError>>,
61        finished: Arc<AtomicBool>,
62    }
63
64    impl<T> JoinHandle<T> {
65        /// Request cancellation without waiting for task cleanup.
66        pub fn abort(&self) {
67            self.abort.abort();
68        }
69
70        /// Whether the task has completed or observed cancellation.
71        #[must_use]
72        pub fn is_finished(&self) -> bool {
73            self.finished.load(Ordering::Acquire)
74        }
75    }
76
77    impl<T> Future for JoinHandle<T> {
78        type Output = Result<T, JoinError>;
79
80        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
81            match Pin::new(&mut self.result).poll(cx) {
82                Poll::Ready(Ok(result)) => Poll::Ready(result),
83                Poll::Ready(Err(_)) => Poll::Ready(Err(JoinError)),
84                Poll::Pending => Poll::Pending,
85            }
86        }
87    }
88
89    impl<T> fmt::Debug for JoinHandle<T> {
90        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
91            f.debug_struct("JoinHandle")
92                .field("finished", &self.is_finished())
93                .finish()
94        }
95    }
96
97    fn parts<T>() -> (
98        futures_util::future::AbortRegistration,
99        oneshot::Sender<Result<T, JoinError>>,
100        JoinHandle<T>,
101        Arc<AtomicBool>,
102    ) {
103        let (abort, registration) = AbortHandle::new_pair();
104        let (send, result) = oneshot::channel();
105        let finished = Arc::new(AtomicBool::new(false));
106        let handle = JoinHandle {
107            abort,
108            result,
109            finished: finished.clone(),
110        };
111        (registration, send, handle, finished)
112    }
113
114    /// Spawn a `Send` task which may follow its waker across browser workers.
115    pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
116    where
117        F: Future + Send + 'static,
118        F::Output: Send + 'static,
119    {
120        let (registration, send, handle, finished) = parts();
121        web_task::spawn(async move {
122            let result = Abortable::new(future, registration)
123                .await
124                .map_err(|_| JoinError);
125            finished.store(true, Ordering::Release);
126            let _ = send.send(result);
127        })
128        .detach();
129        handle
130    }
131
132    /// Spawn a task pinned to the current browser thread.
133    pub fn spawn_local<F>(future: F) -> JoinHandle<F::Output>
134    where
135        F: Future + 'static,
136        F::Output: 'static,
137    {
138        let (registration, send, handle, finished) = parts();
139        web_task::spawn_local(async move {
140            let result = Abortable::new(future, registration)
141                .await
142                .map_err(|_| JoinError);
143            finished.store(true, Ordering::Release);
144            let _ = send.send(result);
145        })
146        .detach();
147        handle
148    }
149}
150
151#[cfg(not(target_arch = "wasm32"))]
152pub use native::*;
153#[cfg(target_arch = "wasm32")]
154pub use web::*;
cross_runtime/task.rs

cross_runtime/
task.rs
