fusor_async/
cancellation.rs

1use futures_util::future::{AbortHandle, Abortable, FutureExt, LocalBoxFuture};
2#[cfg(feature = "browser")]
3use std::cell::OnceCell;
4use std::{
5    cell::{Cell, RefCell},
6    collections::BTreeMap,
7    future::Future,
8    rc::{Rc, Weak},
9};
10
11type Callback = Box<dyn FnOnce()>;
12#[derive(Default)]
13struct Cancellation {
14    cancelled: Cell<bool>,
15    next: Cell<u64>,
16    callbacks: RefCell<BTreeMap<u64, Callback>>,
17    #[cfg(feature = "browser")]
18    controller: OnceCell<web_sys::AbortController>,
19}
20
21/// Cancellation of a single operation. Connect transport cancellation here; stopping
22/// polling alone does not stop browser Fetch or undo a server operation.
23#[derive(Clone, Default)]
24pub struct CancellationToken(Rc<Cancellation>);
25
26/// Owner-side cancellation capability, also usable by explicit write adapters.
27/// Cancellation stops local work; it cannot establish whether a server committed.
28/// Dropping this source cancels its token. Receiving tokens cannot cancel it.
29pub struct CancellationSource(Option<CancellationToken>);
30impl Default for CancellationSource {
31    fn default() -> Self {
32        Self(Some(CancellationToken::default()))
33    }
34}
35impl CancellationSource {
36    pub fn token(&self) -> CancellationToken {
37        self.0.as_ref().expect("live cancellation source").clone()
38    }
39    pub fn cancel(&self) {
40        if let Some(token) = &self.0 {
41            token.cancel();
42        }
43    }
44    /// The operation completed. Release the source without signalling cancellation.
45    pub fn complete(mut self) {
46        self.release();
47    }
48    fn release(&mut self) {
49        self.0.take();
50    }
51}
52impl Drop for CancellationSource {
53    fn drop(&mut self) {
54        self.cancel();
55    }
56}
57
58/// One in-flight load: the handle that stops polling its future and the source
59/// of its token. Dropping it aborts the future, then cancels the token.
60pub(crate) struct InFlight {
61    abort: AbortHandle,
62    source: CancellationSource,
63}
64impl InFlight {
65    /// Build the load from its token. The returned future stops at its next poll
66    /// once this handle is dropped; hand it to the executor unchanged.
67    pub(crate) fn start<F: Future<Output = ()> + 'static>(
68        work: impl FnOnce(CancellationToken) -> F,
69    ) -> (Self, LocalBoxFuture<'static, ()>) {
70        let (abort, registration) = AbortHandle::new_pair();
71        let source = CancellationSource::default();
72        let work = Abortable::new(work(source.token()), registration)
73            .map(drop)
74            .boxed_local();
75        (Self { abort, source }, work)
76    }
77    /// The load finished. Release it without signalling cancellation.
78    pub(crate) fn complete(mut self) {
79        self.source.release();
80    }
81}
82impl Drop for InFlight {
83    fn drop(&mut self) {
84        self.abort.abort();
85    }
86}
87
88/// Unregisters its callback on drop. Keep it alive while the operation is pending.
89#[must_use = "retain the cancellation registration while the operation is pending"]
90pub struct CancelRegistration {
91    token: Weak<Cancellation>,
92    id: u64,
93}
94impl Drop for CancelRegistration {
95    fn drop(&mut self) {
96        if let Some(token) = self.token.upgrade() {
97            let callback = token.callbacks.borrow_mut().remove(&self.id);
98            drop(callback);
99        }
100    }
101}
102impl CancellationToken {
103    pub fn is_cancelled(&self) -> bool {
104        self.0.cancelled.get()
105    }
106    pub fn on_cancel(&self, callback: impl FnOnce() + 'static) -> CancelRegistration {
107        let id = crate::increment(&self.0.next, "cancellation registration");
108        if self.is_cancelled() {
109            callback();
110        } else {
111            self.0.callbacks.borrow_mut().insert(id, Box::new(callback));
112        }
113        CancelRegistration {
114            token: Rc::downgrade(&self.0),
115            id,
116        }
117    }
118    /// A browser `AbortSignal` that aborts when this token is cancelled. Pass it to
119    /// any Fetch; it lives as long as the token, so it also covers body reads.
120    #[cfg(feature = "browser")]
121    pub fn abort_signal(&self) -> Result<web_sys::AbortSignal, wasm_bindgen::JsValue> {
122        if let Some(controller) = self.0.controller.get() {
123            return Ok(controller.signal());
124        }
125        let controller = web_sys::AbortController::new()?;
126        if self.is_cancelled() {
127            controller.abort();
128        }
129        let signal = controller.signal();
130        let _ = self.0.controller.set(controller);
131        Ok(signal)
132    }
133    pub(crate) fn cancel(&self) {
134        if self.0.cancelled.replace(true) {
135            return;
136        }
137        // Stop the transport before running cleanup callbacks.
138        #[cfg(feature = "browser")]
139        if let Some(controller) = self.0.controller.get() {
140            controller.abort();
141        }
142        let callbacks = self.0.callbacks.take();
143        for callback in callbacks.into_values() {
144            callback();
145        }
146    }
147}
fusor_async/cancellation.rs

fusor_async/
cancellation.rs
