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    /// Request cancellation. Calls in flight stop at their next check point.
24    pub fn cancel(&self) {
25        self.flag.store(true, Ordering::SeqCst);
26    }
27    pub fn is_cancelled(&self) -> bool {
28        self.flag.load(Ordering::SeqCst)
29    }
30    /// The underlying flag, for APIs that take `&AtomicBool`.
31    pub fn as_atomic(&self) -> &AtomicBool {
32        &self.flag
33    }
34}
35
36/// Marker carried inside the `io::Error` produced by a cancelled read.
37#[derive(Debug)]
38pub(crate) struct CancelledIo;
39
40impl fmt::Display for CancelledIo {
41    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
42        f.write_str("request cancelled")
43    }
44}
45
46impl std::error::Error for CancelledIo {}
47
48pub(crate) fn cancelled_io() -> io::Error {
49    io::Error::other(CancelledIo)
50}
51
52/// True when `err` came from a cancelled read on a stream returned by this crate.
53pub fn is_cancelled_io(err: &io::Error) -> bool {
54    err.get_ref().is_some_and(|inner| inner.is::<CancelledIo>())
55}
56
57pub(crate) fn is_set(cancel: Option<&AtomicBool>) -> bool {
58    cancel.is_some_and(|flag| flag.load(Ordering::SeqCst))
59}
60
61/// Sleep for `total`, waking every 50 ms to observe cancellation.
62/// Returns `false` when cancelled before the delay elapsed.
63pub(crate) fn sleep_cancellable(total: Duration, cancel: Option<&AtomicBool>) -> bool {
64    let until = Instant::now() + total;
65    loop {
66        if is_set(cancel) {
67            return false;
68        }
69        let now = Instant::now();
70        if now >= until {
71            return true;
72        }
73        std::thread::sleep((until - now).min(Duration::from_millis(50)));
74    }
75}
76
77/// Reader that refuses to read once the flag is set.
78pub(crate) struct CheckedReader<'a> {
79    pub(crate) inner: &'a mut dyn Read,
80    pub(crate) cancel: Option<&'a AtomicBool>,
81}
82
83impl Read for CheckedReader<'_> {
84    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
85        if is_set(self.cancel) {
86            return Err(cancelled_io());
87        }
88        self.inner.read(buf)
89    }
90}
rightkit_http/cancel.rs

rightkit_http/
cancel.rs
