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;
22use url::Url;
23
24use super::{HttpClientError, client::REQUEST_TIMEOUT_MESSAGE, error::transport_error};
25use crate::dst::time::Instant;
26
27/// An HTTP response whose body is consumed incrementally.
28///
29/// One absolute deadline covers response headers and the whole body, including time between
30/// chunk reads. Dropping an unfinished response releases its body and, under simulation,
31/// aborts its owned connection task. No total body limit applies.
32#[derive(Debug)]
33pub struct HttpResponseStream {
34    pub(super) response: http::Response<Incoming>,
35    pub(super) deadline: Option<Instant>,
36    pub(super) url: Option<Url>,
37    #[cfg(all(feature = "simulation", madsim))]
38    pub(super) _connection: super::simulation::Connection,
39}
40
41impl HttpResponseStream {
42    /// Returns the HTTP response status.
43    #[must_use]
44    pub fn status(&self) -> StatusCode {
45        self.response.status()
46    }
47
48    /// Reads the next body chunk, or returns `None` at the end of the body.
49    ///
50    /// # Errors
51    ///
52    /// Returns an error on a transport failure or when the original request deadline expires.
53    pub async fn chunk(&mut self) -> Result<Option<Bytes>, HttpClientError> {
54        read_chunk(self.response.body_mut(), self.deadline)
55            .await
56            .map_err(|e| response_error(e, self.url.as_ref()))
57    }
58}
59
60pub(super) async fn read_chunk<B>(
61    body: &mut B,
62    deadline: Option<Instant>,
63) -> Result<Option<Bytes>, HttpClientError>
64where
65    B: http_body::Body<Data = Bytes> + Unpin,
66    B::Error: std::error::Error + 'static,
67{
68    loop {
69        let frame = match deadline {
70            Some(deadline) => {
71                if Instant::now() >= deadline {
72                    return Err(HttpClientError::TimeoutError(
73                        REQUEST_TIMEOUT_MESSAGE.into(),
74                    ));
75                }
76                tokio::select! {
77                    biased;
78                    () = crate::dst::time::sleep_until(deadline) => return Err(HttpClientError::TimeoutError(REQUEST_TIMEOUT_MESSAGE.into())),
79                    frame = body.frame() => frame,
80                }
81            }
82            None => body.frame().await,
83        };
84        let Some(frame) = frame else {
85            return Ok(None);
86        };
87
88        if let Ok(chunk) = frame.map_err(|e| transport_error(&e))?.into_data() {
89            return Ok(Some(chunk));
90        }
91    }
92}
93
94pub(super) fn response_error(error: HttpClientError, url: Option<&Url>) -> HttpClientError {
95    match (error, url) {
96        (HttpClientError::TransportError(message), Some(url)) => {
97            HttpClientError::TransportError(format!("{message} for url ({url})"))
98        }
99        (error, _) => error,
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use std::io;
106
107    use http::{HeaderMap, HeaderValue, header::HeaderName};
108    use http_body::Frame;
109    use http_body_util::StreamBody;
110    use rstest::rstest;
111
112    use super::*;
113
114    #[tokio::test]
115    async fn read_chunk_skips_trailers_and_preserves_data() {
116        let trailers = HeaderMap::from_iter([(
117            HeaderName::from_static("x-checksum"),
118            HeaderValue::from_static("receipt-83"),
119        )]);
120        let frames: Vec<Result<_, io::Error>> = vec![
121            Ok(Frame::data(Bytes::from_static(b"first"))),
122            Ok(Frame::data(Bytes::from_static(b"second"))),
123            Ok(Frame::trailers(trailers)),
124        ];
125        let mut body = StreamBody::new(futures_util::stream::iter(frames));
126
127        let first = read_chunk(&mut body, None).await.unwrap();
128        let second = read_chunk(&mut body, None).await.unwrap();
129        let end = read_chunk(&mut body, None).await.unwrap();
130
131        assert_eq!(first, Some(Bytes::from_static(b"first")));
132        assert_eq!(second, Some(Bytes::from_static(b"second")));
133        assert_eq!(end, None);
134    }
135
136    #[rstest]
137    #[case::transport(io::ErrorKind::UnexpectedEof, false)]
138    #[case::timeout(io::ErrorKind::TimedOut, true)]
139    #[tokio::test]
140    async fn read_chunk_propagates_body_error(#[case] kind: io::ErrorKind, #[case] timeout: bool) {
141        let frames = vec![Err::<Frame<Bytes>, _>(io::Error::new(kind, "body failure"))];
142        let mut body = StreamBody::new(futures_util::stream::iter(frames));
143
144        let error = read_chunk(&mut body, None).await.unwrap_err();
145
146        match (error, timeout) {
147            (HttpClientError::TimeoutError(message), true)
148            | (HttpClientError::TransportError(message), false) => {
149                assert_eq!(message, "body failure");
150            }
151            (error, _) => panic!("unexpected classification: {error:?}"),
152        }
153    }
154}
nautilus_network/http/stream.rs

nautilus_network/http/
stream.rs
