stream_download/http/
reqwest_client.rs

1//! Adapters for using [`reqwest`] with `stream-download`
2
3use std::str::FromStr;
4use std::sync::LazyLock;
5
6use bytes::Bytes;
7use futures_util::Stream;
8use reqwest::header::{self, AsHeaderName, HeaderMap};
9use tracing::warn;
10
11use super::{DecodeError, RANGE_HEADER_KEY, format_range_header_bytes};
12use crate::http::{Client, ClientResponse, ResponseHeaders};
13
14impl ResponseHeaders for HeaderMap {
15    fn header(&self, name: &str) -> Option<&str> {
16        get_header_str(self, name)
17    }
18}
19
20fn get_header_str<K: AsHeaderName>(headers: &HeaderMap, key: K) -> Option<&str> {
21    headers.get(key).and_then(|val| {
22        val.to_str()
23            .inspect_err(|e| warn!("error converting header value: {e:?}"))
24            .ok()
25    })
26}
27
28/// Error returned when making an HTTP call
29#[derive(thiserror::Error, Debug)]
30#[error("Failed to fetch: {source}")]
31pub struct FetchError {
32    #[source]
33    source: reqwest::Error,
34    // Box to prevent large error variant warnings
35    response: Box<reqwest::Response>,
36}
37
38impl FetchError {
39    /// Error source.
40    pub fn source(&self) -> &reqwest::Error {
41        &self.source
42    }
43
44    /// Http response.
45    pub fn response(&self) -> &reqwest::Response {
46        &self.response
47    }
48}
49
50impl DecodeError for FetchError {
51    async fn decode_error(self) -> String {
52        match self.response.text().await {
53            Ok(text) => format!("{}: {text}", self.source),
54            Err(e) => format!("{}. Error decoding response: {e}", self.source),
55        }
56    }
57}
58
59impl ClientResponse for reqwest::Response {
60    type ResponseError = FetchError;
61    type StreamError = reqwest::Error;
62    type Headers = HeaderMap;
63
64    fn content_length(&self) -> Option<u64> {
65        get_header_str(self.headers(), header::CONTENT_LENGTH).and_then(|content_length| {
66            u64::from_str(content_length)
67                .inspect_err(|e| warn!("invalid content length value: {e:?}"))
68                .ok()
69        })
70    }
71
72    fn content_type(&self) -> Option<&str> {
73        get_header_str(self.headers(), header::CONTENT_TYPE)
74    }
75
76    fn headers(&self) -> Self::Headers {
77        self.headers().clone()
78    }
79
80    fn into_result(self) -> Result<Self, Self::ResponseError> {
81        if let Err(error) = self.error_for_status_ref() {
82            Err(FetchError {
83                source: error,
84                response: Box::new(self),
85            })
86        } else {
87            Ok(self)
88        }
89    }
90
91    fn stream(
92        self,
93    ) -> Box<dyn Stream<Item = Result<Bytes, Self::StreamError>> + Unpin + Send + Sync> {
94        Box::new(self.bytes_stream())
95    }
96}
97
98// per reqwest's docs, it's advisable to create a single client and reuse it
99static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(reqwest::Client::new);
100
101impl Client for reqwest::Client {
102    type Url = reqwest::Url;
103    type Response = reqwest::Response;
104    type Error = reqwest::Error;
105    type Headers = HeaderMap;
106
107    fn create() -> Self {
108        CLIENT.clone()
109    }
110
111    async fn get(&self, url: &Self::Url) -> Result<Self::Response, Self::Error> {
112        self.get(url.clone()).send().await
113    }
114
115    async fn get_range(
116        &self,
117        url: &Self::Url,
118        start: u64,
119        end: Option<u64>,
120    ) -> Result<Self::Response, Self::Error> {
121        self.get(url.clone())
122            .header(RANGE_HEADER_KEY, format_range_header_bytes(start, end))
123            .send()
124            .await
125    }
126}
stream_download/http/reqwest_client.rs

stream_download/http/
reqwest_client.rs
