object_store/client/http/
connection.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
18use crate::ClientOptions;
19#[cfg(feature = "reqwest")]
20use crate::client::HttpResponseBody;
21use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
22use crate::client::{HttpRequest, HttpResponse};
23use async_trait::async_trait;
24use http::{Method, Uri};
25#[cfg(feature = "reqwest")]
26use http_body_util::BodyExt;
27use std::error::Error;
28use std::sync::Arc;
29#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
30use tokio::runtime::Handle;
31
32/// An HTTP protocol error
33///
34/// Clients should return this when an HTTP request fails to be completed, e.g. because
35/// of a connection issue. This does **not** include HTTP requests that are return
36/// non 2xx Status Codes, as these should instead be returned as an [`HttpResponse`]
37/// with the appropriate status code set.
38#[derive(Debug, thiserror::Error)]
39#[error("HTTP error: {source}")]
40pub struct HttpError {
41    kind: HttpErrorKind,
42    #[source]
43    source: Box<dyn Error + Send + Sync>,
44}
45
46/// Identifies the kind of [`HttpError`]
47///
48/// This is used, among other things, to determine if a request can be retried
49#[derive(Debug, Copy, Clone, PartialEq, Eq)]
50#[non_exhaustive]
51pub enum HttpErrorKind {
52    /// An error occurred whilst connecting to the remote
53    ///
54    /// Will be automatically retried
55    Connect,
56    /// An error occurred whilst making the request
57    ///
58    /// Will be automatically retried
59    Request,
60    /// Request timed out
61    ///
62    /// Will be automatically retried if the request is idempotent
63    Timeout,
64    /// The request was aborted
65    ///
66    /// Will be automatically retried if the request is idempotent
67    Interrupted,
68    /// An error occurred whilst decoding the response
69    ///
70    /// Will not be automatically retried
71    Decode,
72    /// An unknown error occurred
73    ///
74    /// Will not be automatically retried
75    Unknown,
76}
77
78impl HttpError {
79    /// Create a new [`HttpError`] with the optional status code
80    pub fn new<E>(kind: HttpErrorKind, e: E) -> Self
81    where
82        E: Error + Send + Sync + 'static,
83    {
84        Self {
85            kind,
86            source: Box::new(e),
87        }
88    }
89
90    #[cfg(feature = "reqwest")]
91    pub(crate) fn reqwest(e: reqwest::Error) -> Self {
92        #[cfg(not(target_arch = "wasm32"))]
93        let is_connect = || e.is_connect();
94        #[cfg(target_arch = "wasm32")]
95        let is_connect = || false;
96
97        let mut kind = if e.is_timeout() {
98            HttpErrorKind::Timeout
99        } else if is_connect() {
100            HttpErrorKind::Connect
101        } else if e.is_decode() {
102            HttpErrorKind::Decode
103        } else {
104            HttpErrorKind::Unknown
105        };
106
107        // Reqwest error variants aren't great, attempt to refine them
108        let mut source = e.source();
109        while kind == HttpErrorKind::Unknown {
110            if let Some(e) = source {
111                if let Some(e) = e.downcast_ref::<hyper::Error>() {
112                    if e.is_closed() || e.is_incomplete_message() || e.is_body_write_aborted() {
113                        kind = HttpErrorKind::Request;
114                    } else if e.is_timeout() {
115                        kind = HttpErrorKind::Timeout;
116                    }
117                }
118                if let Some(e) = e.downcast_ref::<std::io::Error>() {
119                    match e.kind() {
120                        std::io::ErrorKind::TimedOut => kind = HttpErrorKind::Timeout,
121                        std::io::ErrorKind::ConnectionAborted
122                        | std::io::ErrorKind::ConnectionReset
123                        | std::io::ErrorKind::BrokenPipe
124                        | std::io::ErrorKind::UnexpectedEof => kind = HttpErrorKind::Interrupted,
125                        _ => {}
126                    }
127                }
128                source = e.source();
129            } else {
130                break;
131            }
132        }
133        Self {
134            kind,
135            // We strip URL as it will be included by RetryError if not sensitive
136            source: Box::new(e.without_url()),
137        }
138    }
139
140    /// Returns the [`HttpErrorKind`]
141    pub fn kind(&self) -> HttpErrorKind {
142        self.kind
143    }
144}
145
146/// An asynchronous function from a [`HttpRequest`] to a [`HttpResponse`].
147#[async_trait]
148pub trait HttpService: std::fmt::Debug + Send + Sync + 'static {
149    /// Perform [`HttpRequest`] returning [`HttpResponse`]
150    async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError>;
151}
152
153/// An HTTP client
154#[derive(Debug, Clone)]
155pub struct HttpClient(Arc<dyn HttpService>);
156
157impl HttpClient {
158    /// Create a new [`HttpClient`] from an [`HttpService`]
159    pub fn new(service: impl HttpService + 'static) -> Self {
160        Self(Arc::new(service))
161    }
162
163    /// Performs [`HttpRequest`] using this client
164    pub async fn execute(&self, request: HttpRequest) -> Result<HttpResponse, HttpError> {
165        self.0.call(request).await
166    }
167
168    #[allow(unused)]
169    pub(crate) fn get<U>(&self, url: U) -> HttpRequestBuilder
170    where
171        U: TryInto<Uri>,
172        U::Error: Into<RequestBuilderError>,
173    {
174        self.request(Method::GET, url)
175    }
176
177    #[allow(unused)]
178    pub(crate) fn post<U>(&self, url: U) -> HttpRequestBuilder
179    where
180        U: TryInto<Uri>,
181        U::Error: Into<RequestBuilderError>,
182    {
183        self.request(Method::POST, url)
184    }
185
186    #[allow(unused)]
187    pub(crate) fn put<U>(&self, url: U) -> HttpRequestBuilder
188    where
189        U: TryInto<Uri>,
190        U::Error: Into<RequestBuilderError>,
191    {
192        self.request(Method::PUT, url)
193    }
194
195    #[allow(unused)]
196    pub(crate) fn delete<U>(&self, url: U) -> HttpRequestBuilder
197    where
198        U: TryInto<Uri>,
199        U::Error: Into<RequestBuilderError>,
200    {
201        self.request(Method::DELETE, url)
202    }
203
204    pub(crate) fn request<U>(&self, method: Method, url: U) -> HttpRequestBuilder
205    where
206        U: TryInto<Uri>,
207        U::Error: Into<RequestBuilderError>,
208    {
209        HttpRequestBuilder::new(self.clone())
210            .uri(url)
211            .method(method)
212    }
213}
214
215#[async_trait]
216#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
217impl HttpService for reqwest::Client {
218    async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
219        let (parts, body) = req.into_parts();
220
221        let url = parts.uri.to_string().parse().unwrap();
222        let mut req = reqwest::Request::new(parts.method, url);
223        *req.headers_mut() = parts.headers;
224        *req.body_mut() = Some(body.into_reqwest());
225
226        let r = self.execute(req).await.map_err(HttpError::reqwest)?;
227        let res: http::Response<reqwest::Body> = r.into();
228        let (parts, body) = res.into_parts();
229
230        let body = HttpResponseBody::new(body.map_err(HttpError::reqwest));
231        Ok(HttpResponse::from_parts(parts, body))
232    }
233}
234
235#[async_trait]
236#[cfg(all(feature = "reqwest", target_arch = "wasm32", target_os = "unknown"))]
237impl HttpService for reqwest::Client {
238    async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
239        use futures_channel::{mpsc, oneshot};
240        use futures_util::{SinkExt, StreamExt, TryStreamExt};
241        use http_body_util::{Empty, StreamBody};
242        use wasm_bindgen_futures::spawn_local;
243
244        let (parts, body) = req.into_parts();
245        let url = parts.uri.to_string().parse().unwrap();
246        let mut req = reqwest::Request::new(parts.method, url);
247        *req.headers_mut() = parts.headers;
248        *req.body_mut() = Some(body.into_reqwest());
249
250        let (mut tx, rx) = mpsc::channel(1);
251        let (tx_parts, rx_parts) = oneshot::channel();
252        let res_fut = self.execute(req);
253
254        spawn_local(async move {
255            match res_fut.await.map_err(HttpError::reqwest) {
256                Err(err) => {
257                    let _ = tx_parts.send(Err(err));
258                    drop(tx);
259                }
260                Ok(res) => {
261                    let (mut parts, _) = http::Response::new(Empty::<()>::new()).into_parts();
262                    parts.headers = res.headers().clone();
263                    parts.status = res.status();
264                    let _ = tx_parts.send(Ok(parts));
265                    let mut stream = res.bytes_stream().map_err(HttpError::reqwest);
266                    while let Some(chunk) = stream.next().await {
267                        if let Err(_e) = tx.send(chunk).await {
268                            // Disconnected due to a transitive drop of the receiver
269                            break;
270                        }
271                    }
272                }
273            }
274        });
275
276        let parts = rx_parts.await.unwrap()?;
277        let safe_stream = rx.map(|chunk| {
278            let frame = hyper::body::Frame::data(chunk?);
279            Ok(frame)
280        });
281        let body = HttpResponseBody::new(StreamBody::new(safe_stream));
282
283        Ok(HttpResponse::from_parts(parts, body))
284    }
285}
286
287/// A factory for [`HttpClient`]
288pub trait HttpConnector: std::fmt::Debug + Send + Sync + 'static {
289    /// Create a new [`HttpClient`] with the provided [`ClientOptions`]
290    fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient>;
291}
292
293/// [`HttpConnector`] using [`reqwest::Client`]
294#[derive(Debug, Default)]
295#[allow(missing_copy_implementations)]
296#[cfg(all(
297    feature = "reqwest",
298    not(all(target_arch = "wasm32", target_os = "wasi"))
299))]
300pub struct ReqwestConnector {}
301
302#[cfg(all(
303    feature = "reqwest",
304    not(all(target_arch = "wasm32", target_os = "wasi"))
305))]
306impl HttpConnector for ReqwestConnector {
307    fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient> {
308        let client = options.client()?;
309        Ok(HttpClient::new(client))
310    }
311}
312
313/// [`reqwest::Client`] connector that performs all I/O on the provided tokio
314/// [`Runtime`] (thread pool).
315///
316/// This adapter is most useful when you wish to segregate I/O from CPU bound
317/// work that may be happening on the [`Runtime`].
318///
319/// [`Runtime`]: tokio::runtime::Runtime
320///
321/// # Example: Spawning requests on separate runtime
322///
323/// ```
324/// # use std::sync::Arc;
325/// # use tokio::runtime::Runtime;
326/// # use object_store::azure::MicrosoftAzureBuilder;
327/// # use object_store::client::SpawnedReqwestConnector;
328/// # use object_store::ObjectStore;
329/// # fn get_io_runtime() -> Runtime {
330/// #   tokio::runtime::Builder::new_current_thread().build().unwrap()
331/// # }
332/// # fn main() -> Result<(), object_store::Error> {
333/// // create a tokio runtime for I/O.
334/// let io_runtime: Runtime = get_io_runtime();
335/// // configure a store using the runtime.
336/// let handle = io_runtime.handle().clone(); // get a handle to the same runtime
337/// let store: Arc<dyn ObjectStore> = Arc::new(
338///   MicrosoftAzureBuilder::new()
339///     .with_http_connector(SpawnedReqwestConnector::new(handle))
340///     .with_container_name("my_container")
341///     .with_account("my_account")
342///     .build()?
343///  );
344/// // any requests made using store will be spawned on the io_runtime
345/// # Ok(())
346/// # }
347/// ```
348#[derive(Debug)]
349#[allow(missing_copy_implementations)]
350#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
351pub struct SpawnedReqwestConnector {
352    runtime: Handle,
353}
354
355#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
356impl SpawnedReqwestConnector {
357    /// Create a new [`SpawnedReqwestConnector`] with the provided [`Handle`] to
358    /// a tokio [`Runtime`]
359    ///
360    /// [`Runtime`]: tokio::runtime::Runtime
361    pub fn new(runtime: Handle) -> Self {
362        Self { runtime }
363    }
364}
365
366#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
367impl HttpConnector for SpawnedReqwestConnector {
368    fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient> {
369        let spawn_service = super::SpawnService::new(options.client()?, self.runtime.clone());
370        Ok(HttpClient::new(spawn_service))
371    }
372}
373
374#[cfg(all(feature = "reqwest", target_arch = "wasm32", target_os = "wasi"))]
375pub(crate) fn http_connector(
376    custom: Option<Arc<dyn HttpConnector>>,
377) -> crate::Result<Arc<dyn HttpConnector>> {
378    match custom {
379        Some(x) => Ok(x),
380        None => Err(crate::Error::NotSupported {
381            source: "reqwest is not supported on the WASI architecture; \
382                supply a custom HttpConnector via `.with_http_connector(...)`"
383                .to_string()
384                .into(),
385        }),
386    }
387}
388
389#[cfg(all(not(feature = "reqwest"), target_arch = "wasm32", target_os = "wasi"))]
390pub(crate) fn http_connector(
391    custom: Option<Arc<dyn HttpConnector>>,
392) -> crate::Result<Arc<dyn HttpConnector>> {
393    match custom {
394        Some(x) => Ok(x),
395        None => Err(crate::Error::NotSupported {
396            source: "WASI architectures must provide an HttpConnector"
397                .to_string()
398                .into(),
399        }),
400    }
401}
402
403#[cfg(all(
404    feature = "reqwest",
405    not(all(target_arch = "wasm32", target_os = "wasi"))
406))]
407pub(crate) fn http_connector(
408    custom: Option<Arc<dyn HttpConnector>>,
409) -> crate::Result<Arc<dyn HttpConnector>> {
410    match custom {
411        Some(x) => Ok(x),
412        None => Ok(Arc::new(ReqwestConnector {})),
413    }
414}
415
416#[cfg(all(
417    not(feature = "reqwest"),
418    not(all(target_arch = "wasm32", target_os = "wasi"))
419))]
420pub(crate) fn http_connector(
421    custom: Option<Arc<dyn HttpConnector>>,
422) -> crate::Result<Arc<dyn HttpConnector>> {
423    match custom {
424        Some(x) => Ok(x),
425        None => Err(crate::Error::NotSupported {
426            source: "no built-in HTTP transport: enable the `reqwest` feature \
427                or supply a custom HttpConnector via `.with_http_connector(...)`"
428                .to_string()
429                .into(),
430        }),
431    }
432}
object_store/client/http/connection.rs

object_store/client/http/
connection.rs
