abort_signal/
lib.rs

1//! Cooperative cancellation. Dropping a controller does not abort its signals.
2use futures::future::{select, Either};
3use std::{future::Future, pin::pin};
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub struct Aborted;
7impl std::fmt::Display for Aborted {
8    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
9        f.write_str("operation aborted")
10    }
11}
12impl std::error::Error for Aborted {}
13
14#[cfg(not(target_arch = "wasm32"))]
15mod platform {
16    use tokio_util::sync::CancellationToken;
17
18    #[derive(Clone, Debug, Default)]
19    pub struct AbortController(CancellationToken);
20    #[derive(Clone, Debug)]
21    pub struct AbortSignal(CancellationToken);
22
23    impl AbortController {
24        pub fn new() -> Self {
25            Self::default()
26        }
27        pub fn abort(&self) {
28            self.0.cancel();
29        }
30        pub fn signal(&self) -> AbortSignal {
31            AbortSignal(self.0.clone())
32        }
33        /// Parent cancellation reaches this controller; cancelling the child never reaches its parent.
34        pub fn child_of(signal: &AbortSignal) -> Self {
35            Self(signal.0.child_token())
36        }
37    }
38    impl AbortSignal {
39        pub fn aborted(&self) -> bool {
40            self.0.is_cancelled()
41        }
42        pub async fn cancelled(&self) {
43            self.0.cancelled().await
44        }
45    }
46}
47#[cfg(target_arch = "wasm32")]
48mod platform {
49    use futures::channel::oneshot;
50    use wasm_bindgen::{
51        convert::{FromWasmAbi, OptionFromWasmAbi},
52        describe::WasmDescribe,
53    };
54
55    #[derive(Clone, Debug)]
56    pub struct AbortSignal(web_sys::AbortSignal);
57    #[derive(Clone, Debug)]
58    pub struct AbortController(web_sys::AbortController);
59
60    impl AbortController {
61        pub fn new() -> Self {
62            Self(
63                web_sys::AbortController::new()
64                    .expect("AbortController is required on this platform"),
65            )
66        }
67        pub fn abort(&self) {
68            self.0.abort();
69        }
70        pub fn signal(&self) -> AbortSignal {
71            AbortSignal(self.0.signal())
72        }
73    }
74    impl Default for AbortController {
75        fn default() -> Self {
76            Self::new()
77        }
78    }
79    impl From<web_sys::AbortSignal> for AbortSignal {
80        fn from(signal: web_sys::AbortSignal) -> Self {
81            Self(signal)
82        }
83    }
84    impl AbortSignal {
85        pub fn as_web(&self) -> &web_sys::AbortSignal {
86            &self.0
87        }
88        pub fn aborted(&self) -> bool {
89            self.0.aborted()
90        }
91        pub async fn cancelled(&self) {
92            if self.aborted() {
93                return;
94            }
95            let (sender, receiver) = oneshot::channel();
96            // The guard removes the browser listener even when this future is dropped.
97            let _listener = gloo_events::EventListener::once(&self.0, "abort", move |_| {
98                let _ = sender.send(());
99            });
100            let _ = receiver.await;
101        }
102    }
103    impl WasmDescribe for AbortSignal {
104        fn describe() {
105            <web_sys::AbortSignal as WasmDescribe>::describe();
106        }
107    }
108    impl FromWasmAbi for AbortSignal {
109        type Abi = <web_sys::AbortSignal as FromWasmAbi>::Abi;
110        unsafe fn from_abi(value: Self::Abi) -> Self {
111            Self(unsafe { <web_sys::AbortSignal as FromWasmAbi>::from_abi(value) })
112        }
113    }
114    impl OptionFromWasmAbi for AbortSignal {
115        fn is_none(value: &Self::Abi) -> bool {
116            <web_sys::AbortSignal as OptionFromWasmAbi>::is_none(value)
117        }
118    }
119}
120pub use platform::{AbortController, AbortSignal};
121
122impl AbortSignal {
123    /// Stop polling and drop the supplied future when aborted. Does not preempt synchronous work.
124    pub async fn until<T>(&self, future: impl Future<Output = T>) -> Result<T, Aborted> {
125        match select(pin!(self.cancelled()), pin!(future)).await {
126            Either::Left(_) => Err(Aborted),
127            Either::Right((value, _)) => Ok(value),
128        }
129    }
130}
131
132#[cfg(all(test, not(target_arch = "wasm32")))]
133mod tests {
134    use super::*;
135    use futures::{executor::block_on, FutureExt};
136    #[test]
137    fn cancellation_is_shared_idempotent_and_one_way() {
138        let parent = AbortController::default();
139        let child = AbortController::child_of(&parent.signal());
140        let sibling = AbortController::child_of(&parent.signal());
141        child.abort();
142        child.abort();
143        assert!(child.signal().aborted());
144        assert!(!parent.signal().aborted());
145        assert!(!sibling.signal().aborted());
146        parent.abort();
147        assert!(sibling.signal().aborted());
148        assert!(AbortController::child_of(&parent.signal())
149            .signal()
150            .aborted());
151    }
152    #[test]
153    fn precancellation_skips_work_and_dropping_controller_does_not_cancel() {
154        let controller = AbortController::default();
155        controller.abort();
156        let signal = controller.signal();
157        assert_eq!(
158            block_on(signal.until(async { panic!("must not run") })),
159            Err(Aborted)
160        );
161        let signal = AbortController::default().signal();
162        assert!(signal.cancelled().now_or_never().is_none());
163    }
164    #[test]
165    fn wakes_all_waiters_and_drops_non_send_work() {
166        let controller = AbortController::default();
167        let signal = controller.signal();
168        let owned = std::rc::Rc::new(());
169        let work = owned.clone();
170        let mut first = Box::pin(signal.until(async move {
171            let _guard = work;
172            futures::future::pending::<()>().await;
173        }));
174        assert!(first.as_mut().now_or_never().is_none());
175        let mut second = Box::pin(signal.cancelled());
176        assert!(second.as_mut().now_or_never().is_none());
177        controller.abort();
178        assert_eq!(block_on(first), Err(Aborted));
179        block_on(second);
180        assert_eq!(std::rc::Rc::strong_count(&owned), 1);
181    }
182}
abort_signal/lib.rs

abort_signal/
lib.rs
