nautilus_network/http/
stream.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Streaming HTTP responses with the request's original deadline and connection ownership.
17
18use bytes::Bytes;
19use http::StatusCode;
20use http_body_util::BodyExt;
21use hyper::body::Incoming;
22
23use super::{HttpClientError, client::REQUEST_TIMEOUT_MESSAGE, error::transport_error};
24use crate::dst::time::Instant;
25
26/// An HTTP response whose body is consumed incrementally.
27///
28/// One absolute deadline covers response headers and the whole body, including time between
29/// chunk reads. Dropping an unfinished response releases its body and, under simulation,
30/// aborts its owned connection task. No total body limit applies.
31#[derive(Debug)]
32pub struct HttpResponseStream {
33    pub(super) response: http::Response<Incoming>,
34    pub(super) deadline: Option<Instant>,
35    pub(super) origin: Option<String>,
36    #[cfg(all(feature = "simulation", madsim))]
37    pub(super) _connection: super::simulation::Connection,
38}
39
40impl HttpResponseStream {
41    /// Returns the HTTP response status.
42    #[must_use]
43    pub fn status(&self) -> StatusCode {
44        self.response.status()
45    }
46
47    /// Reads the next body chunk, or returns `None` at the end of the body.
48    ///
49    /// # Errors
50    ///
51    /// Returns an error if the transport fails or the original request deadline expires.
52    pub async fn chunk(&mut self) -> Result<Option<Bytes>, HttpClientError> {
53        read_chunk(self.response.body_mut(), self.deadline)
54            .await
55            .map_err(|e| response_error(e, self.origin.as_deref()))
56    }
57}
58
59pub(super) async fn read_chunk<B>(
60    body: &mut B,
61    deadline: Option<Instant>,
62) -> Result<Option<Bytes>, HttpClientError>
63where
64    B: http_body::Body<Data = Bytes> + Unpin,
65    B::Error: std::error::Error + 'static,
66{
67    loop {
68        check_deadline(deadline)?;
69
70        let frame = match deadline {
71            Some(deadline) => {
72                tokio::select! {
73                    biased;
74                    () = crate::dst::time::sleep_until(deadline) => return Err(HttpClientError::TimeoutError(REQUEST_TIMEOUT_MESSAGE.into())),
75                    frame = body.frame() => frame,
76                }
77            }
78            None => body.frame().await,
79        };
80
81        let Some(frame) = frame else {
82            return Ok(None);
83        };
84
85        if let Ok(chunk) = frame.map_err(|e| transport_error(&e))?.into_data() {
86            return Ok(Some(chunk));
87        }
88    }
89}
90
91pub(super) fn check_deadline(deadline: Option<Instant>) -> Result<(), HttpClientError> {
92    if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
93        return Err(HttpClientError::TimeoutError(
94            REQUEST_TIMEOUT_MESSAGE.into(),
95        ));
96    }
97
98    Ok(())
99}
100
101pub(super) fn response_error(error: HttpClientError, origin: Option<&str>) -> HttpClientError {
102    match (error, origin) {
103        (HttpClientError::TransportError(message), Some(origin)) => {
104            HttpClientError::TransportError(format!("{message} for origin ({origin})"))
105        }
106        (error, _) => error,
107    }
108}
109
110#[cfg(test)]
111mod tests {
112    use std::io;
113
114    use http::{HeaderMap, HeaderValue, header::HeaderName};
115    use http_body::Frame;
116    use http_body_util::StreamBody;
117    use rstest::rstest;
118
119    use super::*;
120
121    #[tokio::test]
122    async fn read_chunk_skips_trailers_and_preserves_data() {
123        let trailers = HeaderMap::from_iter([(
124            HeaderName::from_static("x-checksum"),
125            HeaderValue::from_static("receipt-83"),
126        )]);
127        let frames: Vec<Result<_, io::Error>> = vec![
128            Ok(Frame::data(Bytes::from_static(b"first"))),
129            Ok(Frame::data(Bytes::from_static(b"second"))),
130            Ok(Frame::trailers(trailers)),
131        ];
132        let mut body = StreamBody::new(futures_util::stream::iter(frames));
133
134        let first = read_chunk(&mut body, None).await.unwrap();
135        let second = read_chunk(&mut body, None).await.unwrap();
136        let end = read_chunk(&mut body, None).await.unwrap();
137
138        assert_eq!(first, Some(Bytes::from_static(b"first")));
139        assert_eq!(second, Some(Bytes::from_static(b"second")));
140        assert_eq!(end, None);
141    }
142
143    #[rstest]
144    #[case::transport(io::ErrorKind::UnexpectedEof, false)]
145    #[case::timeout(io::ErrorKind::TimedOut, true)]
146    #[tokio::test]
147    async fn read_chunk_propagates_body_error(#[case] kind: io::ErrorKind, #[case] timeout: bool) {
148        let frames = vec![Err::<Frame<Bytes>, _>(io::Error::new(kind, "body failure"))];
149        let mut body = StreamBody::new(futures_util::stream::iter(frames));
150
151        let error = read_chunk(&mut body, None).await.unwrap_err();
152
153        match (error, timeout) {
154            (HttpClientError::TimeoutError(message), true)
155            | (HttpClientError::TransportError(message), false) => {
156                assert_eq!(message, "body failure");
157            }
158            (error, _) => panic!("unexpected classification: {error:?}"),
159        }
160    }
161}
nautilus_network/http/stream.rs

nautilus_network/http/
stream.rs
