rightkit_http/
cancel.rs

1//! Cooperative cancellation shared by buffered calls and streams.
2//!
3//! A [`CancelToken`] is checked between attempts, while sleeping between retries,
4//! and before every body read. It cannot interrupt a blocking socket call that is
5//! already waiting for response headers; that wait is bounded by the response timeout.
6
7use std::fmt;
8use std::io::{self, Read};
9use std::sync::atomic::{AtomicBool, Ordering};
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12
13/// Shared cancel flag. Clones observe the same flag.
14#[derive(Debug, Clone, Default)]
15pub struct CancelToken {
16    flag: Arc<AtomicBool>,
17}
18
19impl CancelToken {
20    pub fn new() -> Self {
21        Self::default()
22    }
23    /// Adopt an existing flag: the token observes (and sets) that same `Arc`, so a caller that
24    /// already owns a cancel flag (for example a task guard) shares it instead of copying it.
25    pub fn from_flag(flag: Arc<AtomicBool>) -> Self {
26        Self { flag }
27    }
28    /// The shared flag itself, to hand to code that owns an `Arc<AtomicBool>`.
29    pub fn flag(&self) -> &Arc<AtomicBool> {
30        &self.flag
31    }
32    /// Request cancellation. Calls in flight stop at their next check point.
33    pub fn cancel(&self) {
34        self.flag.store(true, Ordering::SeqCst);
35    }
36    pub fn is_cancelled(&self) -> bool {
37        self.flag.load(Ordering::SeqCst)
38    }
39    /// The underlying flag, for APIs that take `&AtomicBool`.
40    pub fn as_atomic(&self) -> &AtomicBool {
41        &self.flag
42    }
43}
44
45/// Marker carried inside the `io::Error` produced by a cancelled read.
46#[derive(Debug)]
47pub(crate) struct CancelledIo;
48
49impl fmt::Display for CancelledIo {
50    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51        f.write_str("request cancelled")
52    }
53}
54
55impl std::error::Error for CancelledIo {}
56
57pub(crate) fn cancelled_io() -> io::Error {
58    io::Error::other(CancelledIo)
59}
60
61/// True when `err` came from a cancelled read on a stream returned by this crate.
62pub fn is_cancelled_io(err: &io::Error) -> bool {
63    err.get_ref().is_some_and(|inner| inner.is::<CancelledIo>())
64}
65
66pub(crate) fn is_set(cancel: Option<&AtomicBool>) -> bool {
67    cancel.is_some_and(|flag| flag.load(Ordering::SeqCst))
68}
69
70/// Sleep for `total`, waking every 50 ms to observe cancellation.
71/// Returns `false` when cancelled before the delay elapsed.
72pub(crate) fn sleep_cancellable(total: Duration, cancel: Option<&AtomicBool>) -> bool {
73    let until = Instant::now() + total;
74    loop {
75        if is_set(cancel) {
76            return false;
77        }
78        let now = Instant::now();
79        if now >= until {
80            return true;
81        }
82        std::thread::sleep((until - now).min(Duration::from_millis(50)));
83    }
84}
85
86/// Reader that refuses to read once the flag is set.
87pub(crate) struct CheckedReader<'a> {
88    pub(crate) inner: &'a mut dyn Read,
89    pub(crate) cancel: Option<&'a AtomicBool>,
90}
91
92impl Read for CheckedReader<'_> {
93    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
94        if is_set(self.cancel) {
95            return Err(cancelled_io());
96        }
97        self.inner.read(buf)
98    }
99}
rightkit_http/cancel.rs

rightkit_http/
cancel.rs
