Use existing reqwest client for AWS S3 requests (#5917)

This removes a lot of duplicate client dependency bloat for roughly
equivalent functionality.

Co-authored-by: Mathijs van Veluw <black.dex@gmail.com>

src/config.rs

@@ -1188,6 +1188,9 @@ fn opendal_operator_for_path(path: &str) -> Result<opendal::Operator, Error> {
 
 #[cfg(s3)]
 fn opendal_s3_operator_for_path(path: &str) -> Result<opendal::Operator, Error> {
+    use crate::http_client::aws::AwsReqwestConnector;
+    use aws_config::{default_provider::credentials::DefaultCredentialsChain, provider_config::ProviderConfig};
+
     // This is a custom AWS credential loader that uses the official AWS Rust
     // SDK config crate to load credentials. This ensures maximum compatibility
     // with AWS credential configurations. For example, OpenDAL doesn't support
@@ -1200,12 +1203,19 @@ fn opendal_s3_operator_for_path(path: &str) -> Result<opendal::Operator, Error>
             use aws_credential_types::provider::ProvideCredentials as _;
             use tokio::sync::OnceCell;
 
-            static DEFAULT_CREDENTIAL_CHAIN: OnceCell<
-                aws_config::default_provider::credentials::DefaultCredentialsChain,
-            > = OnceCell::const_new();
+            static DEFAULT_CREDENTIAL_CHAIN: OnceCell<DefaultCredentialsChain> = OnceCell::const_new();
 
             let chain = DEFAULT_CREDENTIAL_CHAIN
-                .get_or_init(|| aws_config::default_provider::credentials::DefaultCredentialsChain::builder().build())
+                .get_or_init(|| {
+                    let reqwest_client = reqwest::Client::builder().build().unwrap();
+                    let connector = AwsReqwestConnector {
+                        client: reqwest_client,
+                    };
+
+                    let conf = ProviderConfig::default().with_http_client(connector);
+
+                    DefaultCredentialsChain::builder().configure(conf).build()
+                })
                 .await;
 
             let creds = chain.provide_credentials().await?;

src/http_client.rs

@@ -244,3 +244,61 @@ impl Resolve for CustomDnsResolver {
         })
     }
 }
+
+#[cfg(s3)]
+pub(crate) mod aws {
+    use aws_smithy_runtime_api::client::{
+        http::{HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpConnector},
+        orchestrator::HttpResponse,
+        result::ConnectorError,
+        runtime_components::RuntimeComponents,
+    };
+    use reqwest::Client;
+
+    // Adapter that wraps reqwest to be compatible with the AWS SDK
+    #[derive(Debug)]
+    pub(crate) struct AwsReqwestConnector {
+        pub(crate) client: Client,
+    }
+
+    impl HttpConnector for AwsReqwestConnector {
+        fn call(&self, request: aws_smithy_runtime_api::client::orchestrator::HttpRequest) -> HttpConnectorFuture {
+            // Convert the AWS-style request to a reqwest request
+            let client = self.client.clone();
+            let future = async move {
+                let method = reqwest::Method::from_bytes(request.method().as_bytes())
+                    .map_err(|e| ConnectorError::user(Box::new(e)))?;
+                let mut req_builder = client.request(method, request.uri().to_string());
+
+                for (name, value) in request.headers() {
+                    req_builder = req_builder.header(name, value);
+                }
+
+                if let Some(body_bytes) = request.body().bytes() {
+                    req_builder = req_builder.body(body_bytes.to_vec());
+                }
+
+                let response = req_builder.send().await.map_err(|e| ConnectorError::io(Box::new(e)))?;
+
+                let status = response.status().into();
+                let bytes = response.bytes().await.map_err(|e| ConnectorError::io(Box::new(e)))?;
+
+                Ok(HttpResponse::new(status, bytes.into()))
+            };
+
+            HttpConnectorFuture::new(Box::pin(future))
+        }
+    }
+
+    impl HttpClient for AwsReqwestConnector {
+        fn http_connector(
+            &self,
+            _settings: &HttpConnectorSettings,
+            _components: &RuntimeComponents,
+        ) -> SharedHttpConnector {
+            SharedHttpConnector::new(AwsReqwestConnector {
+                client: self.client.clone(),
+            })
+        }
+    }
+}