base_http_client/
body.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4use crate::{Error, driver::lock};
5use base_http::body::{BodyError, BodyFrame, BodyReader, ReadableBody};
6use base_scheduler::Cancellation;
7use event_listener::Event;
8use std::sync::{Arc, Mutex};
9
10#[derive(Clone, Copy, PartialEq, Eq)]
11enum State {
12    Reading,
13    Complete,
14    Abandoned,
15}
16
17pub(crate) struct Consumption {
18    state: Mutex<State>,
19    changed: Event,
20    cancellation: Cancellation,
21}
22impl Consumption {
23    pub fn new(cancellation: Cancellation) -> Arc<Self> {
24        Arc::new(Self {
25            state: Mutex::new(State::Reading),
26            changed: Event::new(),
27            cancellation,
28        })
29    }
30    pub(crate) fn complete(&self) {
31        let mut state = lock(&self.state);
32        if *state == State::Reading {
33            *state = State::Complete;
34        }
35        drop(state);
36        self.changed.notify(usize::MAX);
37    }
38    fn abandon(&self) {
39        let mut state = lock(&self.state);
40        if *state != State::Reading {
41            return;
42        }
43        *state = State::Abandoned;
44        drop(state);
45        self.cancellation.cancel();
46        self.changed.notify(usize::MAX);
47    }
48    pub async fn wait(&self) -> Result<(), Error> {
49        loop {
50            event_listener::listener!(self.changed => listener);
51            match *lock(&self.state) {
52                State::Complete => return Ok(()),
53                State::Abandoned => return Err(Error::Abandoned),
54                State::Reading => {}
55            }
56            listener.await;
57        }
58    }
59}
60
61/// Incoming response frames. Completion or abandonment is reported to the
62/// client driver; response reads do not drive request upload or socket I/O.
63///
64/// A pending read can be dropped without losing a frame, so reads can race other futures such as a command channel or timer. See [polling reads alongside other I/O](base_http::body#polling-reads-alongside-other-io).
65pub struct ResponseBody {
66    pub(crate) reader: BodyReader,
67    pub(crate) consumption: Arc<Consumption>,
68    pub(crate) length: Option<u64>,
69    pub(crate) trailers: bool,
70}
71impl ReadableBody for ResponseBody {
72    type Data = Vec<u8>;
73    type Error = BodyError;
74    fn is_ready(&self) -> bool {
75        self.reader.is_ready()
76    }
77    fn is_empty(&self) -> bool {
78        self.reader.is_empty()
79    }
80    fn length(&self) -> Option<u64> {
81        self.length
82    }
83    fn may_have_trailers(&self) -> bool {
84        self.trailers
85    }
86    fn close(&mut self, error: Option<BodyError>) {
87        self.consumption.abandon();
88        self.reader.close(error);
89    }
90    async fn read(&mut self, cancellation: &Cancellation) -> Result<BodyFrame<Vec<u8>>, BodyError> {
91        let frame = self.reader.read(cancellation).await?;
92        if frame.is_end() {
93            self.consumption.complete();
94        }
95        Ok(frame)
96    }
97}
98impl Drop for ResponseBody {
99    fn drop(&mut self) {
100        self.consumption.abandon();
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    use futures_lite::future::block_on;
108
109    #[test]
110    fn abandonment_is_final_and_cancels_the_exchange() {
111        let cancellation = Cancellation::new();
112        let consumption = Consumption::new(cancellation.clone());
113        consumption.abandon();
114        assert!(cancellation.is_cancelled());
115        // A late end of input cannot turn an abandoned body into a reusable exchange.
116        consumption.complete();
117        let error = block_on(consumption.wait()).unwrap_err();
118        assert_eq!(error.to_string(), "HTTP client response abandoned");
119    }
120}
base_http_client/body.rs

base_http_client/
body.rs
