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        let frame = match deadline {
69            Some(deadline) => {
70                if Instant::now() >= deadline {
71                    return Err(HttpClientError::TimeoutError(
72                        REQUEST_TIMEOUT_MESSAGE.into(),
73                    ));
74                }
75                tokio::select! {
76                    biased;
77                    () = crate::dst::time::sleep_until(deadline) => return Err(HttpClientError::TimeoutError(REQUEST_TIMEOUT_MESSAGE.into())),
78                    frame = body.frame() => frame,
79                }
80            }
81            None => body.frame().await,
82        };
83        let Some(frame) = frame else {
84            return Ok(None);
85        };
86
87        if let Ok(chunk) = frame.map_err(|e| transport_error(&e))?.into_data() {
88            return Ok(Some(chunk));
89        }
90    }
91}
92
93pub(super) fn response_error(error: HttpClientError, origin: Option<&str>) -> HttpClientError {
94    match (error, origin) {
95        (HttpClientError::TransportError(message), Some(origin)) => {
96            HttpClientError::TransportError(format!("{message} for origin ({origin})"))
97        }
98        (error, _) => error,
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use std::io;
105
106    use http::{HeaderMap, HeaderValue, header::HeaderName};
107    use http_body::Frame;
108    use http_body_util::StreamBody;
109    use rstest::rstest;
110
111    use super::*;
112
113    #[tokio::test]
114    async fn read_chunk_skips_trailers_and_preserves_data() {
115        let trailers = HeaderMap::from_iter([(
116            HeaderName::from_static("x-checksum"),
117            HeaderValue::from_static("receipt-83"),
118        )]);
119        let frames: Vec<Result<_, io::Error>> = vec![
120            Ok(Frame::data(Bytes::from_static(b"first"))),
121            Ok(Frame::data(Bytes::from_static(b"second"))),
122            Ok(Frame::trailers(trailers)),
123        ];
124        let mut body = StreamBody::new(futures_util::stream::iter(frames));
125
126        let first = read_chunk(&mut body, None).await.unwrap();
127        let second = read_chunk(&mut body, None).await.unwrap();
128        let end = read_chunk(&mut body, None).await.unwrap();
129
130        assert_eq!(first, Some(Bytes::from_static(b"first")));
131        assert_eq!(second, Some(Bytes::from_static(b"second")));
132        assert_eq!(end, None);
133    }
134
135    #[rstest]
136    #[case::transport(io::ErrorKind::UnexpectedEof, false)]
137    #[case::timeout(io::ErrorKind::TimedOut, true)]
138    #[tokio::test]
139    async fn read_chunk_propagates_body_error(#[case] kind: io::ErrorKind, #[case] timeout: bool) {
140        let frames = vec![Err::<Frame<Bytes>, _>(io::Error::new(kind, "body failure"))];
141        let mut body = StreamBody::new(futures_util::stream::iter(frames));
142
143        let error = read_chunk(&mut body, None).await.unwrap_err();
144
145        match (error, timeout) {
146            (HttpClientError::TimeoutError(message), true)
147            | (HttpClientError::TransportError(message), false) => {
148                assert_eq!(message, "body failure");
149            }
150            (error, _) => panic!("unexpected classification: {error:?}"),
151        }
152    }
153}
nautilus_network/http/stream.rs

nautilus_network/http/
stream.rs
