refactor: Migrate http service to context based http client (#5879)

* refactor: Introduce HttpCore for improved service structure

* refactor: Migrate http service to context based http client

Author: miroim

core/src/services/http/backend.rs

@@ -19,21 +19,20 @@ use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::sync::Arc;
 
-use http::header;
-use http::header::IF_MATCH;
-use http::header::IF_NONE_MATCH;
-use http::Request;
 use http::Response;
 use http::StatusCode;
 use log::debug;
 
+use super::core::HttpCore;
 use super::error::parse_error;
 use crate::raw::*;
 use crate::services::HttpConfig;
 use crate::*;
 
 impl Configurator for HttpConfig {
     type Builder = HttpBuilder;
+
+    #[allow(deprecated)]
     fn into_builder(self) -> Self::Builder {
         HttpBuilder {
             config: self,
@@ -47,6 +46,8 @@ impl Configurator for HttpConfig {
 #[derive(Default)]
 pub struct HttpBuilder {
     config: HttpConfig,
+
+    #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
     http_client: Option<HttpClient>,
 }
 
@@ -119,6 +120,8 @@ impl HttpBuilder {
     ///
     /// This API is part of OpenDAL's Raw API. `HttpClient` could be changed
     /// during minor updates.
+    #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
+    #[allow(deprecated)]
     pub fn http_client(mut self, client: HttpClient) -> Self {
         self.http_client = Some(client);
         self
@@ -143,15 +146,6 @@ impl Builder for HttpBuilder {
         let root = normalize_root(&self.config.root.unwrap_or_default());
         debug!("backend use root {}", root);
 
-        let client = if let Some(client) = self.http_client {
-            client
-        } else {
-            HttpClient::new().map_err(|err| {
-                err.with_operation("Builder::build")
-                    .with_context("service", Scheme::Http)
-            })?
-        };
-
         let mut auth = None;
         if let Some(username) = &self.config.username {
             auth = Some(format_authorization_by_basic(
@@ -163,67 +157,66 @@ impl Builder for HttpBuilder {
             auth = Some(format_authorization_by_bearer(token)?)
         }
 
-        Ok(HttpBackend {
-            info: {
-                let ma = AccessorInfo::default();
-                ma.set_scheme(Scheme::Http)
-                    .set_root(&root)
-                    .set_native_capability(Capability {
-                        stat: true,
-                        stat_with_if_match: true,
-                        stat_with_if_none_match: true,
-                        stat_has_cache_control: true,
-                        stat_has_content_length: true,
-                        stat_has_content_type: true,
-                        stat_has_content_encoding: true,
-                        stat_has_content_range: true,
-                        stat_has_etag: true,
-                        stat_has_content_md5: true,
-                        stat_has_last_modified: true,
-                        stat_has_content_disposition: true,
-
-                        read: true,
-
-                        read_with_if_match: true,
-                        read_with_if_none_match: true,
-
-                        presign: auth.is_none(),
-                        presign_read: auth.is_none(),
-                        presign_stat: auth.is_none(),
-
-                        shared: true,
-
-                        ..Default::default()
-                    });
-
-                ma.into()
-            },
+        let info = AccessorInfo::default();
+        info.set_scheme(Scheme::Http)
+            .set_root(&root)
+            .set_native_capability(Capability {
+                stat: true,
+                stat_with_if_match: true,
+                stat_with_if_none_match: true,
+                stat_has_cache_control: true,
+                stat_has_content_length: true,
+                stat_has_content_type: true,
+                stat_has_content_encoding: true,
+                stat_has_content_range: true,
+                stat_has_etag: true,
+                stat_has_content_md5: true,
+                stat_has_last_modified: true,
+                stat_has_content_disposition: true,
+
+                read: true,
+
+                read_with_if_match: true,
+                read_with_if_none_match: true,
+
+                presign: auth.is_none(),
+                presign_read: auth.is_none(),
+                presign_stat: auth.is_none(),
+
+                shared: true,
+
+                ..Default::default()
+            });
+
+        // allow deprecated api here for compatibility
+        #[allow(deprecated)]
+        if let Some(client) = self.http_client {
+            info.update_http_client(|_| client);
+        }
+
+        let accessor_info = Arc::new(info);
+
+        let core = Arc::new(HttpCore {
+            info: accessor_info,
             endpoint: endpoint.to_string(),
-            authorization: auth,
             root,
-            client,
-        })
+            authorization: auth,
+        });
+
+        Ok(HttpBackend { core })
     }
 }
 
 /// Backend is used to serve `Accessor` support for http.
 #[derive(Clone)]
 pub struct HttpBackend {
-    info: Arc<AccessorInfo>,
-
-    endpoint: String,
-    root: String,
-    client: HttpClient,
-
-    authorization: Option<String>,
+    core: Arc<HttpCore>,
 }
 
 impl Debug for HttpBackend {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Backend")
-            .field("endpoint", &self.endpoint)
-            .field("root", &self.root)
-            .field("client", &self.client)
+        f.debug_struct("HttpBackend")
+            .field("core", &self.core)
             .finish()
     }
 }
@@ -239,7 +232,7 @@ impl Access for HttpBackend {
     type BlockingDeleter = ();
 
     fn info(&self) -> Arc<AccessorInfo> {
-        self.info.clone()
+        self.core.info.clone()
     }
 
     async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
@@ -248,7 +241,7 @@ impl Access for HttpBackend {
             return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
         }
 
-        let resp = self.http_head(path, &args).await?;
+        let resp = self.core.http_head(path, &args).await?;
 
         let status = resp.status();
 
@@ -264,7 +257,7 @@ impl Access for HttpBackend {
     }
 
     async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
-        let resp = self.http_get(path, args.range(), &args).await?;
+        let resp = self.core.http_get(path, args.range(), &args).await?;
 
         let status = resp.status();
 
@@ -281,16 +274,18 @@ impl Access for HttpBackend {
     }
 
     async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
-        if self.has_authorization() {
+        if self.core.has_authorization() {
             return Err(Error::new(
                 ErrorKind::Unsupported,
                 "Http doesn't support presigned request on backend with authorization",
             ));
         }
 
         let req = match args.operation() {
-            PresignOperation::Stat(v) => self.http_head_request(path, v)?,
-            PresignOperation::Read(v) => self.http_get_request(path, BytesRange::default(), v)?,
+            PresignOperation::Stat(v) => self.core.http_head_request(path, v)?,
+            PresignOperation::Read(v) => {
+                self.core.http_get_request(path, BytesRange::default(), v)?
+            }
             _ => {
                 return Err(Error::new(
                     ErrorKind::Unsupported,
@@ -308,77 +303,3 @@ impl Access for HttpBackend {
         )))
     }
 }
-
-impl HttpBackend {
-    pub fn has_authorization(&self) -> bool {
-        self.authorization.is_some()
-    }
-
-    pub fn http_get_request(
-        &self,
-        path: &str,
-        range: BytesRange,
-        args: &OpRead,
-    ) -> Result<Request<Buffer>> {
-        let p = build_rooted_abs_path(&self.root, path);
-
-        let url = format!("{}{}", self.endpoint, percent_encode_path(&p));
-
-        let mut req = Request::get(&url);
-
-        if let Some(if_match) = args.if_match() {
-            req = req.header(IF_MATCH, if_match);
-        }
-
-        if let Some(if_none_match) = args.if_none_match() {
-            req = req.header(IF_NONE_MATCH, if_none_match);
-        }
-
-        if let Some(auth) = &self.authorization {
-            req = req.header(header::AUTHORIZATION, auth.clone())
-        }
-
-        if !range.is_full() {
-            req = req.header(header::RANGE, range.to_header());
-        }
-
-        req.body(Buffer::new()).map_err(new_request_build_error)
-    }
-
-    pub async fn http_get(
-        &self,
-        path: &str,
-        range: BytesRange,
-        args: &OpRead,
-    ) -> Result<Response<HttpBody>> {
-        let req = self.http_get_request(path, range, args)?;
-        self.client.fetch(req).await
-    }
-
-    pub fn http_head_request(&self, path: &str, args: &OpStat) -> Result<Request<Buffer>> {
-        let p = build_rooted_abs_path(&self.root, path);
-
-        let url = format!("{}{}", self.endpoint, percent_encode_path(&p));
-
-        let mut req = Request::head(&url);
-
-        if let Some(if_match) = args.if_match() {
-            req = req.header(IF_MATCH, if_match);
-        }
-
-        if let Some(if_none_match) = args.if_none_match() {
-            req = req.header(IF_NONE_MATCH, if_none_match);
-        }
-
-        if let Some(auth) = &self.authorization {
-            req = req.header(header::AUTHORIZATION, auth.clone())
-        }
-
-        req.body(Buffer::new()).map_err(new_request_build_error)
-    }
-
-    async fn http_head(&self, path: &str, args: &OpStat) -> Result<Response<Buffer>> {
-        let req = self.http_head_request(path, args)?;
-        self.client.send(req).await
-    }
-}