opendal_http_transport_reqwest/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22
23use std::fmt::{Debug, Formatter};
24use std::future;
25use std::mem;
26use std::sync::LazyLock;
27
28use futures::TryStreamExt;
29use http::Request;
30use http::Response;
31use opendal_core::Buffer;
32use opendal_core::Error;
33use opendal_core::ErrorKind;
34use opendal_core::HttpBody;
35use opendal_core::HttpRedirect;
36use opendal_core::HttpTransport;
37use opendal_core::HttpTransporter;
38use opendal_core::Result;
39use opendal_core::raw::parse_content_encoding;
40use opendal_core::raw::parse_content_length;
41
42static DEFAULT_REQWEST_TRANSPORT: LazyLock<ReqwestTransport> =
43    LazyLock::new(|| ReqwestTransport::new(reqwest::Client::new()));
44
45/// A HTTP transport with [`reqwest::Client`].
46#[derive(Clone)]
47pub struct ReqwestTransport {
48    client: reqwest::Client,
49}
50
51impl Debug for ReqwestTransport {
52    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
53        f.debug_struct("ReqwestTransport").finish()
54    }
55}
56
57impl Default for ReqwestTransport {
58    fn default() -> Self {
59        DEFAULT_REQWEST_TRANSPORT.clone()
60    }
61}
62
63impl From<reqwest::Client> for ReqwestTransport {
64    fn from(client: reqwest::Client) -> Self {
65        Self::new(client)
66    }
67}
68
69impl ReqwestTransport {
70    /// Create a new transport from a [`reqwest::Client`].
71    pub fn new(client: reqwest::Client) -> Self {
72        Self { client }
73    }
74}
75
76impl HttpTransport for ReqwestTransport {
77    async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
78        // Uri stores all string alike data in `Bytes` which means
79        // the clone here is cheap.
80        let uri = req.uri().clone();
81        let is_head = req.method() == http::Method::HEAD;
82
83        let (mut parts, body) = req.into_parts();
84        let original = uri.to_string();
85        let target = parts
86            .extensions
87            .get::<HttpRedirect>()
88            .map(|redirect| redirect.uri().original_uri())
89            .unwrap_or(&original);
90
91        let url = reqwest::Url::parse(target).map_err(|err| {
92            Error::new(ErrorKind::Unexpected, "request url is invalid")
93                .with_operation("reqwest::fetch")
94                .with_context("url", uri.to_string())
95                .set_source(err)
96        })?;
97
98        if parts.extensions.get::<HttpRedirect>().is_some() {
99            let original = reqwest::Url::parse(&uri.to_string()).map_err(|err| {
100                Error::new(ErrorKind::Unexpected, "original request url is invalid").set_source(err)
101            })?;
102            if original.origin() != url.origin() {
103                // A reused redirect must not gain credentials from client
104                // defaults or cookie storage that a normal redirect strips.
105                for name in [
106                    "authorization",
107                    "proxy-authorization",
108                    "cookie",
109                    "cookie2",
110                    "www-authenticate",
111                ] {
112                    parts.headers.insert(
113                        http::header::HeaderName::from_static(name),
114                        http::HeaderValue::from_static(""),
115                    );
116                }
117            }
118        }
119
120        let mut req_builder = self
121            .client
122            .request(parts.method, url)
123            .headers(parts.headers);
124
125        // Client under wasm doesn't support set version.
126        #[cfg(not(target_arch = "wasm32"))]
127        {
128            req_builder = req_builder.version(parts.version);
129        }
130
131        // Don't set body if body is empty.
132        if !body.is_empty() {
133            #[cfg(not(target_arch = "wasm32"))]
134            {
135                req_builder = req_builder.body(reqwest::Body::wrap(HttpBufferBody(body)))
136            }
137            #[cfg(target_arch = "wasm32")]
138            {
139                req_builder = req_builder.body(reqwest::Body::from(body.to_bytes()))
140            }
141        }
142
143        let mut resp = req_builder.send().await.map_err(|err| {
144            Error::new(ErrorKind::Unexpected, "send http request")
145                .with_operation("reqwest::send")
146                .with_context("url", uri.to_string())
147                .with_temporary(is_temporary_error(&err))
148                .set_source(err.without_url())
149        })?;
150
151        // Get content length from header so that we can check it.
152        //
153        // - If the request method is HEAD, we will ignore content length.
154        // - If response contains content_encoding, we should omit its content length.
155        let content_length = if is_head || parse_content_encoding(resp.headers())?.is_some() {
156            None
157        } else {
158            parse_content_length(resp.headers())?
159        };
160
161        let mut hr = Response::builder()
162            .status(resp.status())
163            // Insert uri into response extension so that we can fetch
164            // it later.
165            .extension(uri.clone());
166
167        // Optional metadata must not turn a successful fetch into an error.
168        if let Ok(target) = resp.url().as_str().parse::<http::Uri>()
169            && target != uri
170        {
171            let redirect = parts
172                .extensions
173                .get::<HttpRedirect>()
174                .filter(|redirect| redirect.uri().original_uri() == resp.url().as_str())
175                .cloned()
176                .unwrap_or_else(|| HttpRedirect::new(target));
177            hr = hr.extension(redirect);
178        }
179
180        // Response builder under wasm doesn't support set version.
181        #[cfg(not(target_arch = "wasm32"))]
182        {
183            hr = hr.version(resp.version());
184        }
185
186        // Swap headers directly instead of copy the entire map.
187        mem::swap(hr.headers_mut().unwrap(), resp.headers_mut());
188
189        let bs = HttpBody::new(
190            resp.bytes_stream()
191                .try_filter(|v| future::ready(!v.is_empty()))
192                .map_ok(Buffer::from)
193                .map_err(move |err| {
194                    Error::new(ErrorKind::Unexpected, "read data from http response")
195                        .with_operation("reqwest::fetch")
196                        .with_context("url", uri.to_string())
197                        .with_temporary(is_temporary_error(&err))
198                        .set_source(err.without_url())
199                }),
200            content_length,
201        );
202
203        let resp = hr.body(bs).expect("response must build succeed");
204        Ok(resp)
205    }
206}
207
208/// Install the process-wide default reqwest transport.
209///
210/// The reqwest client is initialized when the transport handles its first
211/// request.
212#[doc(hidden)]
213pub fn install_default() {
214    HttpTransporter::install_default(LazyReqwestTransport);
215}
216
217struct LazyReqwestTransport;
218
219impl HttpTransport for LazyReqwestTransport {
220    async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
221        DEFAULT_REQWEST_TRANSPORT.fetch(req).await
222    }
223}
224
225#[inline]
226fn is_temporary_error(err: &reqwest::Error) -> bool {
227    // error sending request
228    err.is_request()||
229    // request or response body error
230    err.is_body() ||
231    // error decoding response body, for example, connection reset.
232    err.is_decode()
233}
234
235#[cfg(not(target_arch = "wasm32"))]
236struct HttpBufferBody(Buffer);
237
238#[cfg(not(target_arch = "wasm32"))]
239impl http_body::Body for HttpBufferBody {
240    type Data = bytes::Bytes;
241    type Error = std::convert::Infallible;
242
243    fn poll_frame(
244        mut self: std::pin::Pin<&mut Self>,
245        _: &mut std::task::Context<'_>,
246    ) -> std::task::Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
247        match self.0.next() {
248            Some(bs) => std::task::Poll::Ready(Some(Ok(http_body::Frame::data(bs)))),
249            None => std::task::Poll::Ready(None),
250        }
251    }
252
253    fn is_end_stream(&self) -> bool {
254        self.0.is_empty()
255    }
256
257    fn size_hint(&self) -> http_body::SizeHint {
258        http_body::SizeHint::with_exact(self.0.len() as u64)
259    }
260}
261
262#[cfg(test)]
263mod tests {
264    use super::*;
265
266    #[test]
267    fn test_install_default_is_lazy() {
268        install_default();
269    }
270
271    #[cfg(any(feature = "rustls", feature = "native-tls"))]
272    #[test]
273    fn test_default_transport_succeeds() {
274        let transport = ReqwestTransport::default();
275        assert_eq!(format!("{:?}", transport), "ReqwestTransport");
276    }
277
278    #[test]
279    fn test_from_reqwest_client() {
280        let client = reqwest::Client::new();
281        let transport = ReqwestTransport::from(client);
282        assert_eq!(format!("{:?}", transport), "ReqwestTransport");
283    }
284}
opendal_http_transport_reqwest/lib.rs

opendal_http_transport_reqwest/
lib.rs
