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.
63pub struct ResponseBody {
64    pub(crate) reader: BodyReader,
65    pub(crate) consumption: Arc<Consumption>,
66    pub(crate) length: Option<u64>,
67    pub(crate) trailers: bool,
68}
69impl ReadableBody for ResponseBody {
70    type Data = Vec<u8>;
71    type Error = BodyError;
72    fn is_ready(&self) -> bool {
73        self.reader.is_ready()
74    }
75    fn is_empty(&self) -> bool {
76        self.reader.is_empty()
77    }
78    fn length(&self) -> Option<u64> {
79        self.length
80    }
81    fn may_have_trailers(&self) -> bool {
82        self.trailers
83    }
84    fn close(&mut self, error: Option<BodyError>) {
85        self.consumption.abandon();
86        self.reader.close(error);
87    }
88    async fn read(&mut self, cancellation: &Cancellation) -> Result<BodyFrame<Vec<u8>>, BodyError> {
89        let frame = self.reader.read(cancellation).await?;
90        if frame.is_end() {
91            self.consumption.complete();
92        }
93        Ok(frame)
94    }
95}
96impl Drop for ResponseBody {
97    fn drop(&mut self) {
98        self.consumption.abandon();
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105    use futures_lite::future::block_on;
106
107    #[test]
108    fn abandonment_is_final_and_cancels_the_exchange() {
109        let cancellation = Cancellation::new();
110        let consumption = Consumption::new(cancellation.clone());
111        consumption.abandon();
112        assert!(cancellation.is_cancelled());
113        // A late end of input cannot turn an abandoned body into a reusable exchange.
114        consumption.complete();
115        let error = block_on(consumption.wait()).unwrap_err();
116        assert_eq!(error.to_string(), "HTTP client response abandoned");
117    }
118}
base_http_client/body.rs

base_http_client/
body.rs
