You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/06/16 11:12:59 UTC
(opendal) 01/01: refactor: introduce http transporter api
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/implement-http-transporter
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit bd9a35b6731c190cd47c6e7e1449cf8309cfcafb
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Jun 16 19:12:40 2026 +0800
refactor: introduce http transporter api
---
core/Cargo.lock | 17 +-
core/Cargo.toml | 14 +-
core/core/Cargo.toml | 14 +-
core/core/src/docs/internals/layer.rs | 6 +-
.../core/src/docs/performance/http_optimization.md | 18 +--
core/core/src/layers/simulate.rs | 2 +-
core/core/src/lib.rs | 18 +--
core/core/src/raw/accessor.rs | 37 +++--
core/core/src/raw/http_util/mod.rs | 12 --
core/core/src/raw/layer.rs | 6 +-
.../http_util => types/http_transport}/body.rs | 33 ++--
core/core/src/types/http_transport/mod.rs | 170 ++++++++++++++++++++
core/core/src/types/mod.rs | 5 +
core/core/src/types/operator/operator.rs | 44 +++---
.../hf => http-transports/reqwest}/Cargo.toml | 31 ++--
.../reqwest/src/lib.rs} | 172 +++++----------------
core/layers/concurrent-limit/src/lib.rs | 28 ++--
core/layers/fastmetrics/src/lib.rs | 4 +-
core/layers/foyer/src/lib.rs | 2 +-
core/layers/hotpath/src/lib.rs | 10 +-
core/layers/metrics/src/lib.rs | 5 +-
core/layers/observe-metrics-common/src/lib.rs | 24 +--
core/layers/otelmetrics/src/lib.rs | 5 +-
core/layers/prometheus-client/src/lib.rs | 5 +-
core/layers/prometheus/src/lib.rs | 4 +-
core/layers/timeout/src/lib.rs | 4 +-
core/layers/tracing/src/lib.rs | 10 +-
core/services/aliyun-drive/src/core.rs | 4 +-
core/services/alluxio/src/core.rs | 22 +--
core/services/azblob/src/core.rs | 6 +-
core/services/azdls/src/core.rs | 6 +-
core/services/azfile/src/core.rs | 6 +-
core/services/b2/src/core.rs | 6 +-
core/services/cloudflare-kv/src/core.rs | 2 +-
core/services/cos/src/core.rs | 6 +-
core/services/d1/src/core.rs | 6 +-
core/services/dbfs/src/core.rs | 10 +-
core/services/dbfs/src/writer.rs | 2 +-
core/services/dropbox/src/core.rs | 20 +--
core/services/gcs/src/core.rs | 6 +-
core/services/gdrive/src/core.rs | 24 +--
core/services/gdrive/src/lister.rs | 2 +-
core/services/gdrive/src/path_index.rs | 10 +-
core/services/ghac/src/core.rs | 18 +--
core/services/github/src/core.rs | 4 +-
core/services/hf/Cargo.toml | 7 +-
core/services/hf/src/backend.rs | 8 +-
core/services/hf/src/core.rs | 18 +--
core/services/hf/src/reader.rs | 7 +-
core/services/http/src/core.rs | 4 +-
core/services/ipfs/src/core.rs | 6 +-
core/services/ipmfs/src/core.rs | 12 +-
core/services/koofr/src/core.rs | 8 +-
core/services/lakefs/src/core.rs | 12 +-
core/services/obs/src/core.rs | 6 +-
core/services/onedrive/src/core.rs | 28 ++--
core/services/oss/src/core.rs | 6 +-
core/services/pcloud/src/core.rs | 4 +-
core/services/s3/src/backend.rs | 10 +-
core/services/s3/src/core.rs | 10 +-
core/services/seafile/src/core.rs | 8 +-
core/services/sqlite/src/backend.rs | 2 +-
core/services/swift/src/core.rs | 22 +--
core/services/tos/src/core.rs | 6 +-
core/services/upyun/src/core.rs | 4 +-
core/services/vercel-artifacts/src/core.rs | 6 +-
core/services/vercel-blob/src/core.rs | 4 +-
core/services/webdav/src/core.rs | 18 +--
core/services/webhdfs/src/core.rs | 24 +--
core/services/yandex-disk/src/core.rs | 4 +-
core/src/lib.rs | 14 +-
71 files changed, 607 insertions(+), 511 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index 92d4abd60..f73a3f3a9 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -6312,6 +6312,7 @@ dependencies = [
"log",
"logforth",
"opendal-core",
+ "opendal-http-transport-reqwest",
"opendal-layer-async-backtrace",
"opendal-layer-await-tree",
"opendal-layer-capability-check",
@@ -6445,7 +6446,6 @@ dependencies = [
"bytes",
"futures",
"http 1.4.1",
- "http-body 1.0.1",
"jiff",
"log",
"logforth",
@@ -6457,7 +6457,6 @@ dependencies = [
"quick-xml",
"rand 0.10.1",
"reqsign-core",
- "reqwest",
"serde",
"serde_json",
"sha2 0.11.0",
@@ -6503,6 +6502,18 @@ dependencies = [
"uuid",
]
+[[package]]
+name = "opendal-http-transport-reqwest"
+version = "0.57.0"
+dependencies = [
+ "bytes",
+ "futures",
+ "http 1.4.1",
+ "http-body 1.0.1",
+ "opendal-core",
+ "reqwest",
+]
+
[[package]]
name = "opendal-layer-async-backtrace"
version = "0.57.0"
@@ -7159,8 +7170,8 @@ dependencies = [
"http 1.4.1",
"log",
"opendal-core",
+ "opendal-http-transport-reqwest",
"percent-encoding",
- "reqwest",
"serde",
"serde_json",
"tokio",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index f31b3b755..ed43b35d6 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -24,6 +24,7 @@ members = [
"edge/*",
"examples/*",
"fuzz",
+ "http-transports/*",
"layers/*",
"services/*",
"testkit",
@@ -85,6 +86,7 @@ auto-register-services = ["dep:ctor"]
blocking = ["opendal-core/blocking"]
default = [
"auto-register-services",
+ "http-transport-reqwest",
"reqwest-rustls-tls",
"executors-tokio",
"layers-concurrent-limit",
@@ -93,6 +95,7 @@ default = [
"layers-timeout",
]
executors-tokio = ["opendal-core/executors-tokio"]
+http-transport-reqwest = ["dep:opendal-http-transport-reqwest"]
internal-path-cache = ["opendal-core/internal-path-cache"]
internal-tokio-rt = ["opendal-core/internal-tokio-rt"]
layers-async-backtrace = ["dep:opendal-layer-async-backtrace"]
@@ -119,8 +122,14 @@ layers-tail-cut = ["dep:opendal-layer-tail-cut"]
layers-throttle = ["dep:opendal-layer-throttle"]
layers-timeout = ["dep:opendal-layer-timeout"]
layers-tracing = ["dep:opendal-layer-tracing"]
-reqwest-rustls-no-provider-tls = ["opendal-core/reqwest-rustls-no-provider-tls"]
-reqwest-rustls-tls = ["opendal-core/reqwest-rustls-tls"]
+reqwest-rustls-no-provider-tls = [
+ "http-transport-reqwest",
+ "opendal-http-transport-reqwest/rustls-no-provider",
+]
+reqwest-rustls-tls = [
+ "http-transport-reqwest",
+ "opendal-http-transport-reqwest/rustls",
+]
services-aliyun-drive = ["dep:opendal-service-aliyun-drive"]
services-alluxio = ["dep:opendal-service-alluxio"]
services-azblob = ["dep:opendal-service-azblob"]
@@ -215,6 +224,7 @@ required-features = ["tests"]
[dependencies]
ctor = { workspace = true, optional = true }
opendal-core = { path = "core", version = "0.57.0", default-features = false }
+opendal-http-transport-reqwest = { path = "http-transports/reqwest", version = "0.57.0", optional = true, default-features = false }
opendal-layer-async-backtrace = { path = "layers/async-backtrace", version = "0.57.0", optional = true, default-features = false }
opendal-layer-await-tree = { path = "layers/await-tree", version = "0.57.0", optional = true, default-features = false }
opendal-layer-capability-check = { path = "layers/capability-check", version = "0.57.0", optional = true, default-features = false }
diff --git a/core/core/Cargo.toml b/core/core/Cargo.toml
index 9b6cf7b54..12d4d3f09 100644
--- a/core/core/Cargo.toml
+++ b/core/core/Cargo.toml
@@ -37,15 +37,7 @@ unused_async = "warn"
all-features = true
[features]
-default = ["reqwest-rustls-tls", "executors-tokio"]
-
-# Enable reqwest rustls tls support.
-reqwest-rustls-tls = ["reqwest/rustls"]
-
-# Enable reqwest rustls tls support without selecting a crypto provider.
-# The downstream binary is responsible for installing a `rustls`
-# `CryptoProvider` (e.g. `ring` or `aws-lc-rs`) before issuing requests.
-reqwest-rustls-no-provider-tls = ["reqwest/rustls-no-provider"]
+default = ["executors-tokio"]
# Enable opendal's blocking support.
blocking = ["internal-tokio-rt"]
@@ -73,7 +65,6 @@ base64 = { workspace = true }
bytes = { workspace = true }
futures = { workspace = true, features = ["std", "async-await"] }
http = { workspace = true }
-http-body = "1"
jiff = { version = "0.2.28", features = ["serde"] }
log = { workspace = true }
md-5 = "0.11.0"
@@ -81,9 +72,6 @@ mea = { workspace = true }
percent-encoding = "2"
quick-xml = { workspace = true, features = ["serialize", "overlapped-lists"] }
reqsign-core = { version = "3.0.1", default-features = false }
-reqwest = { version = "0.13.4", features = [
- "stream",
-], default-features = false }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["macros", "io-util"] }
diff --git a/core/core/src/docs/internals/layer.rs b/core/core/src/docs/internals/layer.rs
index 03aa101fd..619d566d2 100644
--- a/core/core/src/docs/internals/layer.rs
+++ b/core/core/src/docs/internals/layer.rs
@@ -33,7 +33,7 @@
//! ```ignore
//! pub trait Layer: Send + Sync + Debug + Unpin + 'static {
//! fn apply_service(&self, srv: Servicer) -> Servicer;
-//! fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher;
+//! fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter;
//! fn apply_execute(&self, srv: Servicer, inner: Executor) -> Executor;
//! }
//! ```
@@ -66,7 +66,7 @@
//! forwarding calls to [`ServiceDyn`], while the wrapper keeps concrete
//! operation body types until it is returned as a [`Servicer`].
//!
-//! Resource-only layers can implement only `apply_http_fetch` or
+//! Resource-only layers can implement only `apply_http_transport` or
//! `apply_execute`. If they do not need the final service stack, they should
//! name that parameter `_srv`. Layers that need consistent policy across planes
//! can implement multiple hooks, for example operation and HTTP concurrency
@@ -76,7 +76,7 @@
//! [`Service`]: crate::raw::Service
//! [`ServiceDyn`]: crate::raw::ServiceDyn
//! [`Servicer`]: crate::raw::Servicer
-//! [`HttpFetcher`]: crate::raw::HttpFetcher
+//! [`HttpTransporter`]: crate::HttpTransporter
//! [`Executor`]: crate::Executor
//! [`OperationContext`]: crate::raw::OperationContext
//! [`Operator`]: crate::Operator
diff --git a/core/core/src/docs/performance/http_optimization.md b/core/core/src/docs/performance/http_optimization.md
index 3fa1445b9..9a420e496 100644
--- a/core/core/src/docs/performance/http_optimization.md
+++ b/core/core/src/docs/performance/http_optimization.md
@@ -1,6 +1,6 @@
# HTTP Optimization
-All OpenDAL HTTP-based storage services use the same [HttpClient][crate::raw::HttpClient] abstraction. This design offers users a unified interface for configuring HTTP clients. The default HTTP client is [reqwest](https://crates.io/crates/reqwest), a popular and widely used HTTP client library in Rust.
+All OpenDAL HTTP-based storage services use the same [HttpTransporter][crate::HttpTransporter] abstraction. This design offers users a unified interface for configuring HTTP transports. The `opendal` facade installs [reqwest](https://crates.io/crates/reqwest) as the default HTTP transport when the `http-transport-reqwest` feature is enabled.
Many of the services supported by OpenDAL are HTTP-based. This guide aims to provide optimization tips for using HTTP-based storage services. While these tips are also applicable to other HTTP clients, the configuration methods may vary.
@@ -25,8 +25,8 @@ let client = reqwest::ClientBuilder::new()
.build()
.expect("http client must be created");
-// Replace the operator's base HTTP client.
-let op = op.http_client(HttpClient::with(client));
+// Replace the operator's base HTTP transport.
+let op = op.http_transport(HttpTransporter::new(ReqwestTransport::new(client)));
```
## DNS Caching
@@ -44,8 +44,8 @@ let client = reqwest::ClientBuilder::new()
.build()
.expect("http client must be created");
-// Replace the operator's base HTTP client.
-let op = op.http_client(HttpClient::with(client));
+// Replace the operator's base HTTP transport.
+let op = op.http_transport(HttpTransporter::new(ReqwestTransport::new(client)));
```
The default DNS cache settings from `hickory_dns` are generally sufficient for most workloads. However, if you have specific requirements—such as sharing the same DNS cache across multiple HTTP clients or configuring the DNS cache size—you can use the `Xuanwo/reqwest-hickory-resolver` crate to set up a custom DNS resolver.
@@ -77,8 +77,8 @@ let client = reqwest::ClientBuilder::new()
.build()
.expect("http client must be created");
-// Replace the operator's base HTTP client.
-let op = op.http_client(HttpClient::with(client));
+// Replace the operator's base HTTP transport.
+let op = op.http_transport(HttpTransporter::new(ReqwestTransport::new(client)));
```
The `ResolverOpts` has many options that can be configured. For a complete list of options, please refer to the [hickory_resolver documentation](https://docs.rs/hickory-resolver/latest/hickory_resolver/config/struct.ResolverOpts.html).
@@ -104,8 +104,8 @@ let client = reqwest::ClientBuilder::new()
.build()
.expect("http client must be created");
-// Replace the operator's base HTTP client.
-let op = op.http_client(HttpClient::with(client));
+// Replace the operator's base HTTP transport.
+let op = op.http_transport(HttpTransporter::new(ReqwestTransport::new(client)));
```
It's also recommended to use opendal's [`TimeoutLayer`][crate::layers::TimeoutLayer] to prevent slow requests hangs forever. This layer will automatically cancel the request if it takes too long to complete.
diff --git a/core/core/src/layers/simulate.rs b/core/core/src/layers/simulate.rs
index 3dfc091b7..48ef26355 100644
--- a/core/core/src/layers/simulate.rs
+++ b/core/core/src/layers/simulate.rs
@@ -783,7 +783,7 @@ mod tests {
}
.into();
let reader = SimulateReader::new(
- OperationContext::new(HttpClient::default(), Executor::default()),
+ OperationContext::new(),
Arc::new(MockService {
capability: Capability::default(),
}),
diff --git a/core/core/src/lib.rs b/core/core/src/lib.rs
index eb0928a1e..29916adcf 100644
--- a/core/core/src/lib.rs
+++ b/core/core/src/lib.rs
@@ -60,11 +60,11 @@
//! The next setup is to compose layers. Layers are modules that provide extra
//! features for every operation. All builtin layers could be found at [`layers`].
//!
-//! Let's use [`layers::CapabilityOverrideLayer`] and a custom [`HttpClient`] as an example.
+//! Let's use [`layers::CapabilityOverrideLayer`] and a custom [`HttpTransporter`] as an example.
//!
//! ```no_run
//! use opendal_core::layers::CapabilityOverrideLayer;
-//! use opendal_core::raw::HttpClient;
+//! use opendal_core::HttpTransporter;
//! use opendal_core::services;
//! use opendal_core::Operator;
//! use opendal_core::Result;
@@ -75,10 +75,10 @@
//! let builder = services::Memory::default();
//!
//! // Init an operator
-//! let client = HttpClient::new()?;
+//! let transport = HttpTransporter::default();
//! let op = Operator::new(builder)?
-//! // Init with custom HTTP client.
-//! .http_client(client)
+//! // Replace the base HTTP transport.
+//! .http_transport(transport)
//! .layer(CapabilityOverrideLayer::new(|cap| cap));
//!
//! Ok(())
@@ -104,7 +104,7 @@
//!
//! ```no_run
//! use opendal_core::options;
-//! use opendal_core::raw::HttpClient;
+//! use opendal_core::HttpTransporter;
//! use opendal_core::services;
//! use opendal_core::Operator;
//! use opendal_core::Result;
@@ -115,10 +115,10 @@
//! let builder = services::Memory::default();
//!
//! // Init an operator
-//! let client = HttpClient::new()?;
+//! let transport = HttpTransporter::default();
//! let op = Operator::new(builder)?
-//! // Init with custom HTTP client.
-//! .http_client(client);
+//! // Replace the base HTTP transport.
+//! .http_transport(transport);
//!
//! // Fetch this file's metadata
//! let meta = op.stat("hello.txt").await?;
diff --git a/core/core/src/raw/accessor.rs b/core/core/src/raw/accessor.rs
index f68f6f821..7b248730f 100644
--- a/core/core/src/raw/accessor.rs
+++ b/core/core/src/raw/accessor.rs
@@ -92,27 +92,32 @@ impl ServiceInfo {
/// Composed resources passed from operator to services and layers.
///
-/// Layers that replace the HTTP client or executor must keep forwarding
+/// Layers that replace the HTTP transport or executor must keep forwarding
/// requests or tasks to the previous value. Otherwise, composed features such as
/// retry, timeout, tracing, and metrics can be bypassed.
#[derive(Clone, Debug)]
pub struct OperationContext {
- http_client: HttpClient,
+ http_transport: HttpTransporter,
executor: Executor,
}
impl OperationContext {
+ /// Create a new operation context with default resources.
+ pub fn new() -> Self {
+ Self::from_parts(HttpTransporter::default(), Executor::default())
+ }
+
/// Create a new operation context from composed resources.
- pub fn new(http_client: HttpClient, executor: Executor) -> Self {
+ pub fn from_parts(http_transport: HttpTransporter, executor: Executor) -> Self {
Self {
- http_client,
+ http_transport,
executor,
}
}
- /// Get the composed HTTP client.
- pub fn http_client(&self) -> &HttpClient {
- &self.http_client
+ /// Get the composed HTTP transport.
+ pub fn http_transport(&self) -> &HttpTransporter {
+ &self.http_transport
}
/// Get the composed executor.
@@ -121,14 +126,14 @@ impl OperationContext {
}
/// Split into composed resources.
- pub fn into_parts(self) -> (HttpClient, Executor) {
- (self.http_client, self.executor)
+ pub fn into_parts(self) -> (HttpTransporter, Executor) {
+ (self.http_transport, self.executor)
}
- /// Return a copy of this context with a different HTTP client.
- pub fn with_http_client(&self, http_client: HttpClient) -> Self {
+ /// Return a copy of this context with a different HTTP transport.
+ pub fn with_http_transport(&self, http_transport: HttpTransporter) -> Self {
Self {
- http_client,
+ http_transport,
executor: self.executor.clone(),
}
}
@@ -136,12 +141,18 @@ impl OperationContext {
/// Return a copy of this context with a different executor.
pub fn with_executor(&self, executor: Executor) -> Self {
Self {
- http_client: self.http_client.clone(),
+ http_transport: self.http_transport.clone(),
executor,
}
}
}
+impl Default for OperationContext {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
/// Underlying trait of all storage services.
///
/// Every storage backend supported by OpenDAL implements [`Service`]. Backends
diff --git a/core/core/src/raw/http_util/mod.rs b/core/core/src/raw/http_util/mod.rs
index ce5cf7cf7..8c0e1fb32 100644
--- a/core/core/src/raw/http_util/mod.rs
+++ b/core/core/src/raw/http_util/mod.rs
@@ -22,18 +22,6 @@
//! This mod is not a part of OpenDAL's public API. We expose them out to make
//! it easier to develop services and layers outside opendal.
-mod client;
-/// temporary client used by several features
-#[allow(unused_imports)]
-pub use client::GLOBAL_REQWEST_CLIENT;
-pub use client::HttpClient;
-pub use client::HttpClientHttpSend;
-pub use client::HttpFetch;
-pub use client::HttpFetcher;
-
-mod body;
-pub use body::HttpBody;
-
mod header;
pub use header::build_header_value;
pub use header::format_authorization_by_basic;
diff --git a/core/core/src/raw/layer.rs b/core/core/src/raw/layer.rs
index 873d088ef..42e401c13 100644
--- a/core/core/src/raw/layer.rs
+++ b/core/core/src/raw/layer.rs
@@ -24,7 +24,7 @@ use crate::*;
///
/// A layer receives the current stack for each plane and returns the stack that
/// should be used by operators built with this layer. Implementations can wrap
-/// the operation service, HTTP fetcher, task executor, or any combination of
+/// the operation service, HTTP transport, task executor, or any combination of
/// them.
///
/// Hooks take `&self`, so layers that keep mutable state must use interior
@@ -39,10 +39,10 @@ pub trait Layer: Send + Sync + Debug + Unpin + 'static {
srv
}
- /// Intercept the HTTP fetch stack.
+ /// Intercept the HTTP transport stack.
///
/// Return `inner` unchanged for layers that do not affect HTTP requests.
- fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
+ fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
let _ = srv;
inner
}
diff --git a/core/core/src/raw/http_util/body.rs b/core/core/src/types/http_transport/body.rs
similarity index 88%
rename from core/core/src/raw/http_util/body.rs
rename to core/core/src/types/http_transport/body.rs
index aa51f0538..525a36e69 100644
--- a/core/core/src/raw/http_util/body.rs
+++ b/core/core/src/types/http_transport/body.rs
@@ -19,15 +19,21 @@ use std::cmp::Ordering;
use futures::Stream;
use futures::StreamExt;
-use oio::ReadStream;
-use crate::raw::*;
-use crate::*;
+use crate::raw::oio;
+use crate::raw::oio::ReadStream;
+use crate::types::Buffer;
+use crate::types::Error;
+use crate::types::ErrorKind;
+use crate::types::Result;
-/// The streaming body that OpenDAL's HttpClient returned.
+/// The streaming body returned by [`HttpTransporter`].
///
-/// We implement [`oio::ReadStream`] for the `HttpBody`. Services can use `HttpBody` as
-/// [`Access::Read`].
+/// We implement [`oio::ReadStream`] for `HttpBody`. Services can use
+/// `HttpBody` as [`Service::Reader`].
+///
+/// [`HttpTransporter`]: super::HttpTransporter
+/// [`Service::Reader`]: crate::raw::Service::Reader
pub struct HttpBody {
#[cfg(not(target_arch = "wasm32"))]
stream: Box<dyn Stream<Item = Result<Buffer>> + Send + Sync + Unpin + 'static>,
@@ -39,12 +45,12 @@ pub struct HttpBody {
/// # Safety
///
-/// HttpBody is `Send` on non wasm32 targets.
+/// `HttpBody` is `Send` on non-wasm targets.
unsafe impl Send for HttpBody {}
/// # Safety
///
-/// HttpBody is sync on non wasm32 targets.
+/// `HttpBody` is `Sync` on non-wasm targets.
unsafe impl Sync for HttpBody {}
impl HttpBody {
@@ -99,7 +105,11 @@ impl HttpBody {
self
}
- /// Check if the consumed data is equal to the expected content length.
+ /// Read all data from the stream.
+ pub async fn to_buffer(&mut self) -> Result<Buffer> {
+ self.read_all().await
+ }
+
#[inline]
fn check(&self) -> Result<()> {
let Some(expect) = self.size else {
@@ -121,11 +131,6 @@ impl HttpBody {
.set_temporary()),
}
}
-
- /// Read all data from the stream.
- pub async fn to_buffer(&mut self) -> Result<Buffer> {
- self.read_all().await
- }
}
impl oio::ReadStream for HttpBody {
diff --git a/core/core/src/types/http_transport/mod.rs b/core/core/src/types/http_transport/mod.rs
new file mode 100644
index 000000000..ad279b1a0
--- /dev/null
+++ b/core/core/src/types/http_transport/mod.rs
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::OnceLock;
+
+use bytes::Bytes;
+use http::Request;
+use http::Response;
+
+use crate::raw::BoxedFuture;
+use crate::raw::oio::ReadStream;
+use crate::types::Buffer;
+use crate::types::Error;
+use crate::types::ErrorKind;
+use crate::types::Result;
+
+mod body;
+pub use body::HttpBody;
+
+static DEFAULT_HTTP_TRANSPORTER: OnceLock<HttpTransporter> = OnceLock::new();
+
+/// HTTP transport used by OpenDAL services.
+///
+/// Implement this trait to provide a custom HTTP backend. A transport must
+/// support 3xx redirection.
+pub trait HttpTransport: Send + Sync + Unpin + 'static {
+ /// Fetch a request and return a streamable [`HttpBody`].
+ #[cfg(not(target_arch = "wasm32"))]
+ fn fetch(
+ &self,
+ req: Request<Buffer>,
+ ) -> impl Future<Output = Result<Response<HttpBody>>> + Send;
+
+ /// Fetch a request and return a streamable [`HttpBody`].
+ #[cfg(target_arch = "wasm32")]
+ fn fetch(&self, req: Request<Buffer>) -> impl Future<Output = Result<Response<HttpBody>>>;
+}
+
+/// Object-safe version of [`HttpTransport`].
+pub(crate) trait HttpTransportDyn: Send + Sync + Unpin + 'static {
+ /// Fetch a request through a boxed future.
+ fn fetch_dyn(&self, req: Request<Buffer>) -> BoxedFuture<'_, Result<Response<HttpBody>>>;
+}
+
+impl<T: HttpTransport + ?Sized> HttpTransportDyn for T {
+ fn fetch_dyn(&self, req: Request<Buffer>) -> BoxedFuture<'_, Result<Response<HttpBody>>> {
+ Box::pin(self.fetch(req))
+ }
+}
+
+/// Type-erased HTTP transport handle.
+#[derive(Clone)]
+pub struct HttpTransporter {
+ inner: Arc<dyn HttpTransportDyn>,
+}
+
+impl Debug for HttpTransporter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("HttpTransporter").finish()
+ }
+}
+
+impl Default for HttpTransporter {
+ fn default() -> Self {
+ Self::new(DefaultHttpTransport)
+ }
+}
+
+impl HttpTransporter {
+ /// Create a new `HttpTransporter` from an [`HttpTransport`].
+ pub fn new(transport: impl HttpTransport) -> Self {
+ Self {
+ inner: Arc::new(transport),
+ }
+ }
+
+ /// Install the process-wide default HTTP transport.
+ ///
+ /// The first installed transport wins. Later calls are ignored.
+ pub fn install_default(transport: impl HttpTransport) {
+ let _ = DEFAULT_HTTP_TRANSPORTER.set(Self::new(transport));
+ }
+
+ /// Send a request and consume the response body into a [`Buffer`].
+ pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
+ let (parts, mut body) = self.fetch(req).await?.into_parts();
+ let buffer = body.read_all().await?;
+ Ok(Response::from_parts(parts, buffer))
+ }
+
+ /// Fetch a request and return a streamable [`HttpBody`].
+ pub async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
+ self.inner.fetch_dyn(req).await
+ }
+}
+
+impl HttpTransport for HttpTransporter {
+ #[cfg(not(target_arch = "wasm32"))]
+ async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
+ HttpTransporter::fetch(self, req).await
+ }
+
+ #[cfg(target_arch = "wasm32")]
+ async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
+ HttpTransporter::fetch(self, req).await
+ }
+}
+
+impl reqsign_core::HttpSend for HttpTransporter {
+ async fn http_send(&self, req: Request<Bytes>) -> reqsign_core::Result<Response<Bytes>> {
+ let req = req.map(Buffer::from);
+ let resp = self.send(req).await.map_err(|err| {
+ let retryable = err.is_temporary();
+ reqsign_core::Error::unexpected("send request via OpenDAL HttpTransporter")
+ .with_source(err)
+ .set_retryable(retryable)
+ })?;
+
+ let (parts, body) = resp.into_parts();
+ Ok(Response::from_parts(parts, body.to_bytes()))
+ }
+}
+
+#[derive(Debug)]
+struct DefaultHttpTransport;
+
+impl HttpTransport for DefaultHttpTransport {
+ #[cfg(not(target_arch = "wasm32"))]
+ async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
+ fetch_with_default_transport(req).await
+ }
+
+ #[cfg(target_arch = "wasm32")]
+ async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
+ fetch_with_default_transport(req).await
+ }
+}
+
+async fn fetch_with_default_transport(req: Request<Buffer>) -> Result<Response<HttpBody>> {
+ match DEFAULT_HTTP_TRANSPORTER.get() {
+ Some(transport) => transport.fetch(req).await,
+ None => Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "default HTTP transport is not installed",
+ )
+ .with_operation("HttpTransporter::default")
+ .with_context(
+ "advice",
+ "call HttpTransporter::install_default before using HTTP services",
+ )),
+ }
+}
diff --git a/core/core/src/types/mod.rs b/core/core/src/types/mod.rs
index 875750ed3..3ad4baa60 100644
--- a/core/core/src/types/mod.rs
+++ b/core/core/src/types/mod.rs
@@ -21,6 +21,11 @@ pub use mode::EntryMode;
mod buffer;
pub use buffer::Buffer;
+mod http_transport;
+pub use http_transport::HttpBody;
+pub use http_transport::HttpTransport;
+pub use http_transport::HttpTransporter;
+
mod entry;
pub use entry::Entry;
diff --git a/core/core/src/types/operator/operator.rs b/core/core/src/types/operator/operator.rs
index faf85355b..849f99fbd 100644
--- a/core/core/src/types/operator/operator.rs
+++ b/core/core/src/types/operator/operator.rs
@@ -66,7 +66,7 @@ use crate::*;
/// add layers or replace providers independently.
///
/// ```
-/// use opendal_core::raw::HttpClient;
+/// use opendal_core::HttpTransporter;
/// use opendal_core::services::Memory;
/// use opendal_core::Operator;
/// use opendal_core::Result;
@@ -74,9 +74,9 @@ use crate::*;
/// async fn test() -> Result<()> {
/// let op: Operator = Operator::new(Memory::default())?;
///
-/// // OpenDAL will replace the default HTTP client now.
-/// let client = HttpClient::new()?;
-/// let op = op.http_client(client);
+/// // OpenDAL will replace the default HTTP transport now.
+/// let transport = HttpTransporter::default();
+/// let op = op.http_transport(transport);
///
/// Ok(())
/// }
@@ -145,7 +145,7 @@ use crate::*;
pub struct Operator {
// Base providers are the bottom slots that layer and resource changes replay from.
base_srv: Servicer,
- base_http_fetch: HttpFetcher,
+ base_http_transport: HttpTransporter,
base_executor: Executor,
// Layers are the replayable program shared by the service, HTTP, and executor planes.
layers: Arc<[Arc<dyn Layer>]>,
@@ -176,24 +176,26 @@ impl Operator {
/// resources from another layer program.
pub(crate) fn with_parts(
base_srv: Servicer,
- base_http_fetch: HttpFetcher,
+ base_http_transport: HttpTransporter,
base_executor: Executor,
layers: Arc<[Arc<dyn Layer>]>,
) -> Self {
let srv = layers
.iter()
.fold(base_srv.clone(), |srv, layer| layer.apply_service(srv));
- let http_fetch = layers.iter().fold(base_http_fetch.clone(), |inner, layer| {
- layer.apply_http_fetch(srv.clone(), inner)
- });
+ let http_transport = layers
+ .iter()
+ .fold(base_http_transport.clone(), |inner, layer| {
+ layer.apply_http_transport(srv.clone(), inner)
+ });
let executor = layers.iter().fold(base_executor.clone(), |inner, layer| {
layer.apply_execute(srv.clone(), inner)
});
- let ctx = OperationContext::new(HttpClient::with(http_fetch), executor);
+ let ctx = OperationContext::from_parts(http_transport, executor);
Self {
base_srv,
- base_http_fetch,
+ base_http_transport,
base_executor,
layers,
srv,
@@ -225,7 +227,7 @@ impl Operator {
pub fn from_service(srv: Servicer) -> Self {
Self::with_parts(
srv,
- HttpClient::default().into_inner(),
+ HttpTransporter::default(),
Executor::default(),
Arc::from([]),
)
@@ -269,19 +271,19 @@ impl Operator {
/// Replace the base HTTP transport and rebuild composed state.
#[must_use]
- pub fn http_client(self, client: HttpClient) -> Self {
- Self::with_parts(
- self.base_srv,
- client.into_inner(),
- self.base_executor,
- self.layers,
- )
+ pub fn http_transport(self, transport: HttpTransporter) -> Self {
+ Self::with_parts(self.base_srv, transport, self.base_executor, self.layers)
}
/// Replace the base executor and rebuild composed state.
#[must_use]
pub fn executor(self, executor: Executor) -> Self {
- Self::with_parts(self.base_srv, self.base_http_fetch, executor, self.layers)
+ Self::with_parts(
+ self.base_srv,
+ self.base_http_transport,
+ executor,
+ self.layers,
+ )
}
/// Apply a layer to this operator and rebuild composed state.
@@ -291,7 +293,7 @@ impl Operator {
layers.push(Arc::new(layer) as Arc<dyn Layer>);
Self::with_parts(
self.base_srv,
- self.base_http_fetch,
+ self.base_http_transport,
self.base_executor,
Arc::from(layers),
)
diff --git a/core/services/hf/Cargo.toml b/core/http-transports/reqwest/Cargo.toml
similarity index 66%
copy from core/services/hf/Cargo.toml
copy to core/http-transports/reqwest/Cargo.toml
index b0d45cf12..1b72144e2 100644
--- a/core/services/hf/Cargo.toml
+++ b/core/http-transports/reqwest/Cargo.toml
@@ -16,8 +16,8 @@
# under the License.
[package]
-description = "Apache OpenDAL Hugging Face service implementation"
-name = "opendal-service-hf"
+description = "Apache OpenDAL reqwest HTTP transport implementation"
+name = "opendal-http-transport-reqwest"
authors = { workspace = true }
edition = { workspace = true }
@@ -30,24 +30,17 @@ version = { workspace = true }
[package.metadata.docs.rs]
all-features = true
+[features]
+default = []
+rustls = ["reqwest/rustls"]
+rustls-no-provider = ["reqwest/rustls-no-provider"]
+
[dependencies]
bytes = { workspace = true }
-hf-xet = "1.5.0"
+futures = { workspace = true, features = ["std"] }
http = { workspace = true }
-log = { workspace = true }
+http-body = "1"
opendal-core = { path = "../../core", version = "0.57.0", default-features = false }
-percent-encoding = "2"
-reqwest = { version = "0.13.4", default-features = false, features = [
- "rustls",
-] }
-serde = { workspace = true, features = ["derive"] }
-serde_json = { workspace = true }
-
-[dev-dependencies]
-base64 = { workspace = true }
-futures = { workspace = true }
-opendal-core = { path = "../../core", version = "0.57.0", features = [
- "reqwest-rustls-tls",
-] }
-serde_json = { workspace = true }
-tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+reqwest = { version = "0.13.4", features = [
+ "stream",
+], default-features = false }
diff --git a/core/core/src/raw/http_util/client.rs b/core/http-transports/reqwest/src/lib.rs
similarity index 51%
rename from core/core/src/raw/http_util/client.rs
rename to core/http-transports/reqwest/src/lib.rs
index d7359ca25..96fe39f7d 100644
--- a/core/core/src/raw/http_util/client.rs
+++ b/core/http-transports/reqwest/src/lib.rs
@@ -15,173 +15,74 @@
// specific language governing permissions and limitations
// under the License.
+//! Reqwest based HTTP transport for Apache OpenDAL.
+
+#![deny(missing_docs)]
+
use std::convert::Infallible;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::future;
use std::mem;
-use std::ops::Deref;
use std::pin::Pin;
use std::str::FromStr;
-use std::sync::Arc;
use std::sync::LazyLock;
use std::task::Context;
use std::task::Poll;
-use crate::raw::oio::ReadStream;
use bytes::Bytes;
-use futures::Future;
use futures::TryStreamExt;
use http::Request;
use http::Response;
use http_body::Frame;
use http_body::SizeHint;
-
-use super::HttpBody;
-use super::parse_content_encoding;
-use super::parse_content_length;
-use crate::raw::*;
-use crate::*;
-
-/// Http client used across opendal for loading credentials.
-/// This is merely a temporary solution because reqsign requires a reqwest client to be passed.
-/// We will remove it after the next major version of reqsign, which will enable users to provide their own client.
-#[allow(dead_code)]
-pub static GLOBAL_REQWEST_CLIENT: LazyLock<reqwest::Client> = LazyLock::new(reqwest::Client::new);
-
-/// HttpFetcher is a type erased [`HttpFetch`].
-pub type HttpFetcher = Arc<dyn HttpFetchDyn>;
-
-/// An HTTP client instance for OpenDAL's services.
+use opendal_core::Buffer;
+use opendal_core::Error;
+use opendal_core::ErrorKind;
+use opendal_core::HttpBody;
+use opendal_core::HttpTransport;
+use opendal_core::Result;
+use opendal_core::raw::parse_content_encoding;
+use opendal_core::raw::parse_content_length;
+
+static DEFAULT_REQWEST_CLIENT: LazyLock<reqwest::Client> = LazyLock::new(reqwest::Client::new);
+
+/// A [`reqwest::Client`] backed HTTP transport.
///
/// # Notes
///
-/// * A http client must support redirections that follows 3xx response.
+/// Reqwest must be configured with a TLS feature before sending HTTPS requests.
#[derive(Clone)]
-pub struct HttpClient {
- fetcher: HttpFetcher,
-}
-
-/// A reqsign `HttpSend` implementation backed by an OpenDAL HTTP client.
-#[derive(Clone)]
-pub struct HttpClientHttpSend {
- client: HttpClient,
-}
-
-impl HttpClientHttpSend {
- /// Create a new [`HttpClientHttpSend`].
- pub fn new(client: HttpClient) -> Self {
- Self { client }
- }
-}
-
-/// We don't want users to know details about our clients.
-impl Debug for HttpClient {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("HttpClient").finish()
- }
+pub struct ReqwestTransport {
+ client: reqwest::Client,
}
-impl Debug for HttpClientHttpSend {
+impl Debug for ReqwestTransport {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("HttpClientHttpSend").finish()
+ f.debug_struct("ReqwestTransport").finish()
}
}
-impl Default for HttpClient {
+impl Default for ReqwestTransport {
fn default() -> Self {
- Self {
- fetcher: Arc::new(GLOBAL_REQWEST_CLIENT.clone()),
- }
+ Self::new(DEFAULT_REQWEST_CLIENT.clone())
}
}
-impl HttpClient {
- /// Create a new http client in async context.
- pub fn new() -> Result<Self> {
- Ok(Self::default())
- }
-
- /// Construct `Self` with given [`reqwest::Client`]
- pub fn with(client: impl HttpFetch) -> Self {
- let fetcher = Arc::new(client);
- Self { fetcher }
- }
-
- /// Get the inner http client.
- pub fn into_inner(self) -> HttpFetcher {
- self.fetcher
- }
-
- /// Send a request and consume response.
- pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
- let (parts, mut body) = self.fetch(req).await?.into_parts();
- let buffer = body.read_all().await?;
- Ok(Response::from_parts(parts, buffer))
- }
-
- /// Fetch a request and return a streamable [`HttpBody`].
- ///
- /// Services can use [`HttpBody`] as [`Access::Read`].
- pub async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
- self.fetcher.fetch(req).await
+impl From<reqwest::Client> for ReqwestTransport {
+ fn from(client: reqwest::Client) -> Self {
+ Self::new(client)
}
}
-impl reqsign_core::HttpSend for HttpClient {
- async fn http_send(&self, req: Request<Bytes>) -> reqsign_core::Result<Response<Bytes>> {
- let req = req.map(Buffer::from);
- let resp = self.send(req).await.map_err(|err| {
- let retryable = err.is_temporary();
- reqsign_core::Error::unexpected("send request via OpenDAL HttpClient")
- .with_source(err)
- .set_retryable(retryable)
- })?;
-
- let (parts, body) = resp.into_parts();
- Ok(Response::from_parts(parts, body.to_bytes()))
- }
-}
-
-impl reqsign_core::HttpSend for HttpClientHttpSend {
- async fn http_send(&self, req: Request<Bytes>) -> reqsign_core::Result<Response<Bytes>> {
- reqsign_core::HttpSend::http_send(&self.client, req).await
- }
-}
-
-/// HttpFetch is the trait to fetch a request in async way.
-/// User should implement this trait to provide their own http client.
-pub trait HttpFetch: Send + Sync + Unpin + 'static {
- /// Fetch a request in async way.
- fn fetch(
- &self,
- req: Request<Buffer>,
- ) -> impl Future<Output = Result<Response<HttpBody>>> + MaybeSend;
-}
-
-/// HttpFetchDyn is the dyn version of [`HttpFetch`]
-/// which make it possible to use as `Arc<dyn HttpFetchDyn>`.
-/// User should never implement this trait, but use `HttpFetch` instead.
-pub trait HttpFetchDyn: Send + Sync + Unpin + 'static {
- /// The dyn version of [`HttpFetch::fetch`].
- ///
- /// This function returns a boxed future to make it object safe.
- fn fetch_dyn(&self, req: Request<Buffer>) -> BoxedFuture<'_, Result<Response<HttpBody>>>;
-}
-
-impl<T: HttpFetch + ?Sized> HttpFetchDyn for T {
- fn fetch_dyn(&self, req: Request<Buffer>) -> BoxedFuture<'_, Result<Response<HttpBody>>> {
- Box::pin(self.fetch(req))
- }
-}
-
-impl<T: HttpFetchDyn + ?Sized> HttpFetch for Arc<T> {
- async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
- self.deref().fetch_dyn(req).await
+impl ReqwestTransport {
+ /// Create a new transport from a [`reqwest::Client`].
+ pub fn new(client: reqwest::Client) -> Self {
+ Self { client }
}
}
-impl HttpFetch for reqwest::Client {
+impl HttpTransport for ReqwestTransport {
async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
// Uri stores all string alike data in `Bytes` which means
// the clone here is cheap.
@@ -192,12 +93,15 @@ impl HttpFetch for reqwest::Client {
let url = reqwest::Url::from_str(&uri.to_string()).map_err(|err| {
Error::new(ErrorKind::Unexpected, "request url is invalid")
- .with_operation("http_util::Client::send::fetch")
+ .with_operation("reqwest::fetch")
.with_context("url", uri.to_string())
.set_source(err)
})?;
- let mut req_builder = self.request(parts.method, url).headers(parts.headers);
+ let mut req_builder = self
+ .client
+ .request(parts.method, url)
+ .headers(parts.headers);
// Client under wasm doesn't support set version.
#[cfg(not(target_arch = "wasm32"))]
@@ -219,7 +123,7 @@ impl HttpFetch for reqwest::Client {
let mut resp = req_builder.send().await.map_err(|err| {
Error::new(ErrorKind::Unexpected, "send http request")
- .with_operation("http_util::Client::send")
+ .with_operation("reqwest::send")
.with_context("url", uri.to_string())
.with_temporary(is_temporary_error(&err))
.set_source(err)
@@ -256,7 +160,7 @@ impl HttpFetch for reqwest::Client {
.map_ok(Buffer::from)
.map_err(move |err| {
Error::new(ErrorKind::Unexpected, "read data from http response")
- .with_operation("http_util::Client::send")
+ .with_operation("reqwest::fetch")
.with_context("url", uri.to_string())
.with_temporary(is_temporary_error(&err))
.set_source(err)
diff --git a/core/layers/concurrent-limit/src/lib.rs b/core/layers/concurrent-limit/src/lib.rs
index 132896b6c..61a70026e 100644
--- a/core/layers/concurrent-limit/src/lib.rs
+++ b/core/layers/concurrent-limit/src/lib.rs
@@ -165,10 +165,10 @@ where
Arc::new(self.layer(inner))
}
- fn apply_http_fetch(&self, _srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
- // Wrap the current HTTP fetcher so HTTP permits are held until the
+ fn apply_http_transport(&self, _srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
+ // Wrap the current HTTP transport so HTTP permits are held until the
// response body is dropped.
- Arc::new(ConcurrentLimitHttpFetcher::<S> {
+ HttpTransporter::new(ConcurrentLimitHttpTransport::<S> {
inner,
http_semaphore: self.http_semaphore.clone(),
})
@@ -188,12 +188,12 @@ where
}
#[doc(hidden)]
-pub struct ConcurrentLimitHttpFetcher<S: ConcurrentLimitSemaphore> {
- inner: HttpFetcher,
+pub struct ConcurrentLimitHttpTransport<S: ConcurrentLimitSemaphore> {
+ inner: HttpTransporter,
http_semaphore: Option<S>,
}
-impl<S: ConcurrentLimitSemaphore> HttpFetch for ConcurrentLimitHttpFetcher<S>
+impl<S: ConcurrentLimitSemaphore> HttpTransport for ConcurrentLimitHttpTransport<S>
where
S::Permit: Unpin,
{
@@ -840,9 +840,9 @@ mod tests {
async fn concurrent_chunked_read_with_http_limit() {
use opendal_core::raw::*;
- struct EchoFetcher;
+ struct EchoTransport;
- impl HttpFetch for EchoFetcher {
+ impl HttpTransport for EchoTransport {
async fn fetch(&self, req: http::Request<Buffer>) -> Result<http::Response<HttpBody>> {
let data = req.into_body();
let len = data.len() as u64;
@@ -886,7 +886,7 @@ mod tests {
None => backend.content.slice(start..),
};
let req = http::Request::get("http://fake").body(data).unwrap();
- let resp = self.ctx.http_client().fetch(req).await?;
+ let resp = self.ctx.http_transport().fetch(req).await?;
let rp = RpRead::new(Metadata::new(EntryMode::FILE).with_content_length(0));
let stream = resp.into_body();
@@ -1012,7 +1012,7 @@ mod tests {
},
content: content.clone(),
}))
- .http_client(HttpClient::with(EchoFetcher))
+ .http_transport(HttpTransporter::new(EchoTransport))
.layer(ConcurrentLimitLayer::new(1024).with_http_concurrent_limit(2));
// chunk=256 ⇒ 16 HTTP requests, concurrent=4, but only 2 HTTP permits.
@@ -1035,9 +1035,9 @@ mod tests {
#[tokio::test]
async fn http_semaphore_holds_until_body_dropped() {
- struct DummyFetcher;
+ struct DummyTransport;
- impl HttpFetch for DummyFetcher {
+ impl HttpTransport for DummyTransport {
async fn fetch(&self, _req: http::Request<Buffer>) -> Result<Response<HttpBody>> {
let body = HttpBody::new(stream::empty(), None);
Ok(Response::builder()
@@ -1049,8 +1049,8 @@ mod tests {
let semaphore = Arc::new(Semaphore::new(1));
let layer = ConcurrentLimitLayer::new(1).with_http_semaphore(semaphore.clone());
- let fetcher = ConcurrentLimitHttpFetcher::<Arc<Semaphore>> {
- inner: HttpClient::with(DummyFetcher).into_inner(),
+ let fetcher = ConcurrentLimitHttpTransport::<Arc<Semaphore>> {
+ inner: HttpTransporter::new(DummyTransport),
http_semaphore: layer.http_semaphore.clone(),
};
diff --git a/core/layers/fastmetrics/src/lib.rs b/core/layers/fastmetrics/src/lib.rs
index 3f584ae0a..45ba1396c 100644
--- a/core/layers/fastmetrics/src/lib.rs
+++ b/core/layers/fastmetrics/src/lib.rs
@@ -150,8 +150,8 @@ impl Layer for FastmetricsLayer {
}
// Reuse the same interceptor because this layer registers both operation and HTTP metrics.
- fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
- observe::MetricsLayer::new(self.interceptor.clone()).apply_http_fetch(srv, inner)
+ fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
+ observe::MetricsLayer::new(self.interceptor.clone()).apply_http_transport(srv, inner)
}
}
diff --git a/core/layers/foyer/src/lib.rs b/core/layers/foyer/src/lib.rs
index f51649764..02d500587 100644
--- a/core/layers/foyer/src/lib.rs
+++ b/core/layers/foyer/src/lib.rs
@@ -494,7 +494,7 @@ mod tests {
}
fn service_context(_: &Servicer) -> OperationContext {
- OperationContext::new(HttpClient::default(), Executor::default())
+ OperationContext::new()
}
#[tokio::test]
diff --git a/core/layers/hotpath/src/lib.rs b/core/layers/hotpath/src/lib.rs
index 370398e16..8a9e13666 100644
--- a/core/layers/hotpath/src/lib.rs
+++ b/core/layers/hotpath/src/lib.rs
@@ -90,8 +90,8 @@ impl Layer for HotpathLayer {
Arc::new(self.layer(inner))
}
- fn apply_http_fetch(&self, _srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
- Arc::new(HotpathHttpFetcher { inner })
+ fn apply_http_transport(&self, _srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
+ HttpTransporter::new(HotpathHttpTransport { inner })
}
}
@@ -259,11 +259,11 @@ impl<C: oio::Copy> oio::Copy for HotpathWrapper<C> {
}
}
-struct HotpathHttpFetcher {
- inner: HttpFetcher,
+struct HotpathHttpTransport {
+ inner: HttpTransporter,
}
-impl HttpFetch for HotpathHttpFetcher {
+impl HttpTransport for HotpathHttpTransport {
async fn fetch(&self, req: http::Request<Buffer>) -> Result<http::Response<HttpBody>> {
let resp = hotpath::measure_async(LABEL_HTTP_FETCH, self.inner.fetch(req)).await?;
let (parts, body) = resp.into_parts();
diff --git a/core/layers/metrics/src/lib.rs b/core/layers/metrics/src/lib.rs
index f9733b678..9ec197cde 100644
--- a/core/layers/metrics/src/lib.rs
+++ b/core/layers/metrics/src/lib.rs
@@ -24,6 +24,7 @@ use metrics::Label;
use metrics::counter;
use metrics::gauge;
use metrics::histogram;
+use opendal_core::HttpTransporter;
use opendal_core::raw::*;
use opendal_layer_observe_metrics_common as observe;
@@ -96,9 +97,9 @@ impl Layer for MetricsLayer {
observe::MetricsLayer::new(interceptor).apply_service(inner)
}
- fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
+ fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
let interceptor = MetricsInterceptor {};
- observe::MetricsLayer::new(interceptor).apply_http_fetch(srv, inner)
+ observe::MetricsLayer::new(interceptor).apply_http_transport(srv, inner)
}
}
diff --git a/core/layers/observe-metrics-common/src/lib.rs b/core/layers/observe-metrics-common/src/lib.rs
index f0a393b68..129fe00a3 100644
--- a/core/layers/observe-metrics-common/src/lib.rs
+++ b/core/layers/observe-metrics-common/src/lib.rs
@@ -459,9 +459,9 @@ impl<I: MetricsIntercept> Layer for MetricsLayer<I> {
Arc::new(self.layer(inner))
}
- fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
+ fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
// HTTP metrics share the same interceptor and service identity as operation metrics.
- Arc::new(MetricsHttpFetcher {
+ HttpTransporter::new(MetricsHttpTransport {
inner,
info: srv.info(),
interceptor: self.interceptor.clone(),
@@ -480,8 +480,8 @@ impl<I: MetricsIntercept> MetricsLayer<I> {
}
}
-struct MetricsHttpFetcher<I: MetricsIntercept> {
- inner: HttpFetcher,
+struct MetricsHttpTransport<I: MetricsIntercept> {
+ inner: HttpTransporter,
info: ServiceInfo,
interceptor: I,
}
@@ -578,7 +578,7 @@ impl<I: MetricsIntercept> Drop for ExecutingGuard<I> {
}
}
-impl<I: MetricsIntercept> HttpFetch for MetricsHttpFetcher<I> {
+impl<I: MetricsIntercept> HttpTransport for MetricsHttpTransport<I> {
async fn fetch(&self, req: http::Request<Buffer>) -> Result<http::Response<HttpBody>> {
let mut labels = MetricLabels::new(
self.info.clone(),
@@ -1412,11 +1412,11 @@ mod tests {
StreamBody(Vec<Result<Buffer>>),
}
- struct MockHttpFetch {
+ struct MockHttpTransport {
behavior: Mutex<MockFetchBehavior>,
}
- impl HttpFetch for MockHttpFetch {
+ impl HttpTransport for MockHttpTransport {
async fn fetch(&self, _req: http::Request<Buffer>) -> Result<http::Response<HttpBody>> {
let behavior = std::mem::replace(
&mut *self.behavior.lock().unwrap(),
@@ -1446,12 +1446,12 @@ mod tests {
fn build_metrics_http_fetcher(
mock: MockInterceptor,
behavior: MockFetchBehavior,
- ) -> MetricsHttpFetcher<MockInterceptor> {
- let inner_fetch = MockHttpFetch {
+ ) -> MetricsHttpTransport<MockInterceptor> {
+ let inner_fetch = MockHttpTransport {
behavior: Mutex::new(behavior),
};
- MetricsHttpFetcher {
- inner: Arc::new(inner_fetch) as HttpFetcher,
+ MetricsHttpTransport {
+ inner: HttpTransporter::new(inner_fetch),
info: test_info(),
interceptor: mock,
}
@@ -1529,7 +1529,7 @@ mod tests {
}
#[tokio::test]
- async fn test_http_client_error_records_status_error() {
+ async fn test_http_transport_error_records_status_error() {
let mock = MockInterceptor::default();
let fetcher = build_metrics_http_fetcher(
mock.clone(),
diff --git a/core/layers/otelmetrics/src/lib.rs b/core/layers/otelmetrics/src/lib.rs
index f806f5b8b..2fb310384 100644
--- a/core/layers/otelmetrics/src/lib.rs
+++ b/core/layers/otelmetrics/src/lib.rs
@@ -20,6 +20,7 @@
#![cfg_attr(docsrs, feature(doc_cfg))]
#![deny(missing_docs)]
+use opendal_core::HttpTransporter;
use opendal_core::raw::*;
use opendal_layer_observe_metrics_common as observe;
use opentelemetry::KeyValue;
@@ -342,8 +343,8 @@ impl Layer for OtelMetricsLayer {
observe::MetricsLayer::new(self.interceptor.clone()).apply_service(inner)
}
- fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
- observe::MetricsLayer::new(self.interceptor.clone()).apply_http_fetch(srv, inner)
+ fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
+ observe::MetricsLayer::new(self.interceptor.clone()).apply_http_transport(srv, inner)
}
}
diff --git a/core/layers/prometheus-client/src/lib.rs b/core/layers/prometheus-client/src/lib.rs
index 4a6ae1143..13447aa6d 100644
--- a/core/layers/prometheus-client/src/lib.rs
+++ b/core/layers/prometheus-client/src/lib.rs
@@ -20,6 +20,7 @@
#![cfg_attr(docsrs, feature(doc_cfg))]
#![deny(missing_docs)]
+use opendal_core::HttpTransporter;
use opendal_core::raw::*;
use opendal_layer_observe_metrics_common as observe;
use prometheus_client::encoding::EncodeLabel;
@@ -93,8 +94,8 @@ impl Layer for PrometheusClientLayer {
}
// Reuse the same interceptor because this layer registers both operation and HTTP metrics.
- fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
- observe::MetricsLayer::new(self.interceptor.clone()).apply_http_fetch(srv, inner)
+ fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
+ observe::MetricsLayer::new(self.interceptor.clone()).apply_http_transport(srv, inner)
}
}
diff --git a/core/layers/prometheus/src/lib.rs b/core/layers/prometheus/src/lib.rs
index 92852d733..495865bcf 100644
--- a/core/layers/prometheus/src/lib.rs
+++ b/core/layers/prometheus/src/lib.rs
@@ -183,8 +183,8 @@ impl Layer for PrometheusLayer {
}
// Reuse the same interceptor because this layer registers both operation and HTTP metrics.
- fn apply_http_fetch(&self, srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
- observe::MetricsLayer::new(self.interceptor.clone()).apply_http_fetch(srv, inner)
+ fn apply_http_transport(&self, srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
+ observe::MetricsLayer::new(self.interceptor.clone()).apply_http_transport(srv, inner)
}
}
diff --git a/core/layers/timeout/src/lib.rs b/core/layers/timeout/src/lib.rs
index 3d55ca642..4e24b526f 100644
--- a/core/layers/timeout/src/lib.rs
+++ b/core/layers/timeout/src/lib.rs
@@ -647,7 +647,7 @@ mod tests {
let service = TimeoutLayer::default()
.with_io_timeout(Duration::from_millis(100))
.apply_service(Arc::new(MockService));
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
let mut copier = service
.copy(&ctx, "f", "t", OpCopy::default(), OpCopier::default())
.unwrap();
@@ -664,7 +664,7 @@ mod tests {
.with_timeout(Duration::from_secs(1))
.with_io_timeout(Duration::from_secs(1));
let service = timeout_layer.apply_service(Arc::new(MockService));
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
let mut lister = service.list(&ctx, "test", OpList::default()).unwrap();
diff --git a/core/layers/tracing/src/lib.rs b/core/layers/tracing/src/lib.rs
index 8f0576150..79dba2e26 100644
--- a/core/layers/tracing/src/lib.rs
+++ b/core/layers/tracing/src/lib.rs
@@ -144,9 +144,9 @@ impl Layer for TracingLayer {
Arc::new(self.layer(inner))
}
- fn apply_http_fetch(&self, _srv: Servicer, inner: HttpFetcher) -> HttpFetcher {
+ fn apply_http_transport(&self, _srv: Servicer, inner: HttpTransporter) -> HttpTransporter {
// Give outbound HTTP requests and their response bodies dedicated spans.
- Arc::new(TracingHttpFetcher { inner })
+ HttpTransporter::new(TracingHttpTransport { inner })
}
}
@@ -156,11 +156,11 @@ impl TracingLayer {
}
}
-struct TracingHttpFetcher {
- inner: HttpFetcher,
+struct TracingHttpTransport {
+ inner: HttpTransporter,
}
-impl HttpFetch for TracingHttpFetcher {
+impl HttpTransport for TracingHttpTransport {
async fn fetch(&self, req: http::Request<Buffer>) -> Result<http::Response<HttpBody>> {
let span = span!(Level::DEBUG, "http::fetch", ?req);
diff --git a/core/services/aliyun-drive/src/core.rs b/core/services/aliyun-drive/src/core.rs
index ee2a3fda3..5f834702d 100644
--- a/core/services/aliyun-drive/src/core.rs
+++ b/core/services/aliyun-drive/src/core.rs
@@ -106,7 +106,7 @@ impl AliyunDriveCore {
.expect("access token must be valid header value"),
);
}
- let res = ctx.http_client().send(req).await?;
+ let res = ctx.http_transport().send(req).await?;
if !res.status().is_success() {
return Err(parse_error(res));
}
@@ -335,7 +335,7 @@ impl AliyunDriveCore {
.header(header::RANGE, range.to_header())
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn move_path(
diff --git a/core/services/alluxio/src/core.rs b/core/services/alluxio/src/core.rs
index 6beb5d1b4..e3996e037 100644
--- a/core/services/alluxio/src/core.rs
+++ b/core/services/alluxio/src/core.rs
@@ -76,7 +76,7 @@ impl AlluxioCore {
.body(Buffer::from(body))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
@@ -110,7 +110,7 @@ impl AlluxioCore {
.body(Buffer::from(body))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
@@ -138,7 +138,7 @@ impl AlluxioCore {
.extension(ServiceOperation("OpenFile"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -167,7 +167,7 @@ impl AlluxioCore {
.extension(ServiceOperation("Delete"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -200,7 +200,7 @@ impl AlluxioCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -225,7 +225,7 @@ impl AlluxioCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -259,7 +259,7 @@ impl AlluxioCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -299,7 +299,7 @@ impl AlluxioCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub(super) async fn write(
@@ -319,7 +319,7 @@ impl AlluxioCore {
let req = req.body(body).map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -346,7 +346,7 @@ impl AlluxioCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -370,7 +370,7 @@ mod tests {
endpoint: "http://127.0.0.1:1".to_string(),
};
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
let err = match core.read(&ctx, 1, BytesRange::from(0_u64..1)).await {
Ok(_) => panic!("range read should be rejected"),
Err(err) => err,
diff --git a/core/services/azblob/src/core.rs b/core/services/azblob/src/core.rs
index f293e84ff..56117069a 100644
--- a/core/services/azblob/src/core.rs
+++ b/core/services/azblob/src/core.rs
@@ -95,7 +95,7 @@ impl AzblobCore {
self.signer.clone().with_context(
Context::new()
.with_file_read(reqsign_file_read_tokio::TokioFileRead)
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone()))
+ .with_http_send(ctx.http_transport().clone())
.with_env(reqsign_core::OsEnv),
)
}
@@ -167,7 +167,7 @@ impl AzblobCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub fn insert_sse_headers(&self, mut req: http::request::Builder) -> http::request::Builder {
@@ -271,7 +271,7 @@ impl AzblobCore {
let req = self.azblob_get_blob_request(path, range, args)?;
let req = self.sign(ctx, req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub fn azblob_put_blob_request(
diff --git a/core/services/azdls/src/core.rs b/core/services/azdls/src/core.rs
index 1647d95fb..f72682d7b 100644
--- a/core/services/azdls/src/core.rs
+++ b/core/services/azdls/src/core.rs
@@ -77,7 +77,7 @@ impl AzdlsCore {
self.signer.clone().with_context(
self.sign_ctx
.clone()
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone())),
+ .with_http_send(ctx.http_transport().clone()),
)
}
@@ -109,7 +109,7 @@ impl AzdlsCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
@@ -156,7 +156,7 @@ impl AzdlsCore {
.map_err(new_request_build_error)?;
let req = self.sign(ctx, req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
/// resource should be one of `file` or `directory`
diff --git a/core/services/azfile/src/core.rs b/core/services/azfile/src/core.rs
index 32e6c35c7..b1aa78ec4 100644
--- a/core/services/azfile/src/core.rs
+++ b/core/services/azfile/src/core.rs
@@ -67,7 +67,7 @@ impl AzfileCore {
self.signer.clone().with_context(
self.sign_ctx
.clone()
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone())),
+ .with_http_send(ctx.http_transport().clone()),
)
}
@@ -95,7 +95,7 @@ impl AzfileCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn azfile_read(
@@ -125,7 +125,7 @@ impl AzfileCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
let req = self.sign(ctx, req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn azfile_create_file(
diff --git a/core/services/b2/src/core.rs b/core/services/b2/src/core.rs
index 262e4a2ce..b6499e9d8 100644
--- a/core/services/b2/src/core.rs
+++ b/core/services/b2/src/core.rs
@@ -74,7 +74,7 @@ impl B2Core {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// [b2_authorize_account](https://www.backblaze.com/apidocs/b2-authorize-account)
@@ -103,7 +103,7 @@ impl B2Core {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
@@ -163,7 +163,7 @@ impl B2Core {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub(super) async fn get_upload_url(
diff --git a/core/services/cloudflare-kv/src/core.rs b/core/services/cloudflare-kv/src/core.rs
index d826c6364..4b106acf8 100644
--- a/core/services/cloudflare-kv/src/core.rs
+++ b/core/services/cloudflare-kv/src/core.rs
@@ -39,7 +39,7 @@ pub struct CloudflareKvCore {
impl CloudflareKvCore {
#[inline]
async fn send(&self, ctx: &OperationContext, req: Request<Buffer>) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
fn sign(&self, req: request::Builder) -> request::Builder {
diff --git a/core/services/cos/src/core.rs b/core/services/cos/src/core.rs
index edbd33cdf..943cf910a 100644
--- a/core/services/cos/src/core.rs
+++ b/core/services/cos/src/core.rs
@@ -70,7 +70,7 @@ impl CosCore {
self.signer.clone().with_context(
Context::new()
.with_file_read(reqsign_file_read_tokio::TokioFileRead)
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone()))
+ .with_http_send(ctx.http_transport().clone())
.with_env(reqsign_core::OsEnv),
)
}
@@ -108,7 +108,7 @@ impl CosCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
@@ -123,7 +123,7 @@ impl CosCore {
let req = self.cos_get_object_request(path, range, args)?;
let req = self.sign(ctx, req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub fn cos_get_object_request(
diff --git a/core/services/d1/src/core.rs b/core/services/d1/src/core.rs
index 249a0f0ff..bdead81db 100644
--- a/core/services/d1/src/core.rs
+++ b/core/services/d1/src/core.rs
@@ -92,7 +92,7 @@ impl D1Core {
let req =
self.create_d1_query_request(&query, vec![path.into()], Operation::Read, "Get")?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
@@ -119,7 +119,7 @@ impl D1Core {
let params = vec![path.into(), value.to_vec().into()];
let req = self.create_d1_query_request(&query, params, Operation::Write, "Set")?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok(()),
@@ -132,7 +132,7 @@ impl D1Core {
let req =
self.create_d1_query_request(&query, vec![path.into()], Operation::Delete, "Delete")?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok(()),
diff --git a/core/services/dbfs/src/core.rs b/core/services/dbfs/src/core.rs
index c34d8de78..b41952f5e 100644
--- a/core/services/dbfs/src/core.rs
+++ b/core/services/dbfs/src/core.rs
@@ -73,7 +73,7 @@ impl DbfsCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn dbfs_delete(
@@ -105,7 +105,7 @@ impl DbfsCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn dbfs_rename(
@@ -136,7 +136,7 @@ impl DbfsCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn dbfs_list(&self, ctx: &OperationContext, path: &str) -> Result<Response<Buffer>> {
@@ -160,7 +160,7 @@ impl DbfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub fn dbfs_create_file_request(&self, path: &str, body: Bytes) -> Result<Request<Buffer>> {
@@ -212,7 +212,7 @@ impl DbfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn dbfs_ensure_parent_path(&self, ctx: &OperationContext, path: &str) -> Result<()> {
diff --git a/core/services/dbfs/src/writer.rs b/core/services/dbfs/src/writer.rs
index eab5c323a..5ede83a81 100644
--- a/core/services/dbfs/src/writer.rs
+++ b/core/services/dbfs/src/writer.rs
@@ -54,7 +54,7 @@ impl oio::OneShotWrite for DbfsWriter {
.core
.dbfs_create_file_request(&self.path, bs.to_bytes())?;
- let resp = self.ctx.http_client().send(req).await?;
+ let resp = self.ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
diff --git a/core/services/dropbox/src/core.rs b/core/services/dropbox/src/core.rs
index e468cbc9d..dcfa97ea6 100644
--- a/core/services/dropbox/src/core.rs
+++ b/core/services/dropbox/src/core.rs
@@ -84,7 +84,7 @@ impl DropboxCore {
.body(Buffer::from(bs))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(request).await?;
+ let resp = ctx.http_transport().send(request).await?;
let body = resp.into_body();
let token: DropboxTokenResponse =
@@ -134,7 +134,7 @@ impl DropboxCore {
let mut request = req.body(Buffer::new()).map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- ctx.http_client().fetch(request).await
+ ctx.http_transport().fetch(request).await
}
pub async fn dropbox_update(
@@ -172,7 +172,7 @@ impl DropboxCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
pub async fn dropbox_delete(
@@ -196,7 +196,7 @@ impl DropboxCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
pub async fn dropbox_create_folder(
@@ -220,7 +220,7 @@ impl DropboxCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- let resp = ctx.http_client().send(request).await?;
+ let resp = ctx.http_transport().send(request).await?;
let status = resp.status();
match status {
StatusCode::OK => Ok(RpCreateDir::default()),
@@ -262,7 +262,7 @@ impl DropboxCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
pub async fn dropbox_list_continue(
@@ -287,7 +287,7 @@ impl DropboxCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
pub async fn dropbox_copy(
@@ -314,7 +314,7 @@ impl DropboxCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
pub async fn dropbox_move(
@@ -341,7 +341,7 @@ impl DropboxCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
pub async fn dropbox_get_metadata(
@@ -367,7 +367,7 @@ impl DropboxCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
}
diff --git a/core/services/gcs/src/core.rs b/core/services/gcs/src/core.rs
index 0bcbcc486..dbbc7ef99 100644
--- a/core/services/gcs/src/core.rs
+++ b/core/services/gcs/src/core.rs
@@ -88,7 +88,7 @@ impl GcsCore {
self.signer.clone().with_context(
self.sign_ctx
.clone()
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone())),
+ .with_http_send(ctx.http_transport().clone()),
)
}
@@ -149,7 +149,7 @@ impl GcsCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
@@ -240,7 +240,7 @@ impl GcsCore {
let req = self.gcs_get_object_request(path, range, args)?;
let req = self.sign(ctx, req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub fn gcs_insert_object_request(
diff --git a/core/services/gdrive/src/core.rs b/core/services/gdrive/src/core.rs
index 70052ee1d..c0ffcb337 100644
--- a/core/services/gdrive/src/core.rs
+++ b/core/services/gdrive/src/core.rs
@@ -251,7 +251,7 @@ impl GdriveCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn gdrive_get(
@@ -293,7 +293,7 @@ impl GdriveCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn gdrive_list(
@@ -323,7 +323,7 @@ impl GdriveCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// List multiple directories in a single API call using OR query.
@@ -371,7 +371,7 @@ impl GdriveCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
// Update with content and metadata
@@ -418,7 +418,7 @@ impl GdriveCore {
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn gdrive_trash(
@@ -441,7 +441,7 @@ impl GdriveCore {
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Create a file with the content.
@@ -491,7 +491,7 @@ impl GdriveCore {
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Overwrite the file with the content.
@@ -520,7 +520,7 @@ impl GdriveCore {
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn sign<T>(&self, ctx: &OperationContext, req: &mut Request<T>) -> Result<()> {
@@ -570,7 +570,7 @@ impl GdriveCore {
.map_err(new_request_build_error)?;
self.sign(ctx, &mut req).await?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
@@ -752,7 +752,7 @@ impl GdriveSigner {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
@@ -830,7 +830,7 @@ impl crate::path_index::GdrivePathQueryer for GdrivePathQuery {
self.signer.lock().await.sign(ctx, &mut req).await?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
@@ -874,7 +874,7 @@ impl crate::path_index::GdrivePathQueryer for GdrivePathQuery {
self.signer.lock().await.sign(ctx, &mut req).await?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
if !resp.status().is_success() {
return Err(parse_error(resp));
}
diff --git a/core/services/gdrive/src/lister.rs b/core/services/gdrive/src/lister.rs
index ed1a9f0c9..a308ea78b 100644
--- a/core/services/gdrive/src/lister.rs
+++ b/core/services/gdrive/src/lister.rs
@@ -797,7 +797,7 @@ mod tests {
}
fn mock_operation_context() -> OperationContext {
- OperationContext::new(HttpClient::default(), Executor::default())
+ OperationContext::new()
}
#[tokio::test]
diff --git a/core/services/gdrive/src/path_index.rs b/core/services/gdrive/src/path_index.rs
index 331a3159b..2d488d8c2 100644
--- a/core/services/gdrive/src/path_index.rs
+++ b/core/services/gdrive/src/path_index.rs
@@ -399,7 +399,7 @@ mod tests {
#[tokio::test]
async fn test_upsert_replaces_existing_path() {
let index = GdrivePathIndex::new(TestQuery::new());
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
index.upsert_file("root/file", "old-id").await;
index.upsert_file("root/file", "new-id").await;
@@ -412,7 +412,7 @@ mod tests {
#[tokio::test]
async fn test_dir_upsert_replaces_file_mapping() {
let index = GdrivePathIndex::new(TestQuery::new());
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
index.upsert_file("root/dir", "file-id").await;
index.upsert_dir("root/dir", "dir-id").await;
@@ -434,7 +434,7 @@ mod tests {
query.insert("remote-root", "root/", "svc-root").await;
query.insert("svc-root", "dir/", "dir-old").await;
query.insert("dir-old", "file", "file-old").await;
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
assert_eq!(
index.get(&ctx, "root/dir/file").await.unwrap().as_deref(),
@@ -459,7 +459,7 @@ mod tests {
let index = GdrivePathIndex::new(query.clone());
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
let (first, second) = tokio::join!(
index.ensure_dir(&ctx, "root/dir"),
index.ensure_dir(&ctx, "root/dir")
@@ -476,7 +476,7 @@ mod tests {
let index = GdrivePathIndex::new(query.clone());
index.upsert_dir("root/", "root-old").await;
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
query.mark_stale_parent("root-old").await;
query.insert("remote-root", "root/", "root-new").await;
diff --git a/core/services/ghac/src/core.rs b/core/services/ghac/src/core.rs
index ec4b637f5..57ae12beb 100644
--- a/core/services/ghac/src/core.rs
+++ b/core/services/ghac/src/core.rs
@@ -122,7 +122,7 @@ impl GhacCore {
.extension(ServiceOperation("GetCacheEntry"))
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let location = if resp.status() == StatusCode::OK {
let slc = resp.into_body();
let query_resp: GhacQueryResponse = serde_json::from_reader(slc.reader())
@@ -157,7 +157,7 @@ impl GhacCore {
.extension(ServiceOperation("GetCacheEntryDownloadURL"))
.body(body)
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let location = if resp.status() == StatusCode::OK {
let slc = resp.into_body();
let query_resp = ghac_types::GetCacheEntryDownloadUrlResponse::decode(slc)
@@ -199,7 +199,7 @@ impl GhacCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn ghac_read(
@@ -221,7 +221,7 @@ impl GhacCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn ghac_get_upload_url(&self, ctx: &OperationContext, path: &str) -> Result<String> {
@@ -248,7 +248,7 @@ impl GhacCore {
.extension(ServiceOperation("ReserveCache"))
.body(Buffer::from(Bytes::from(bs)))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let cache_id = if resp.status().is_success() {
let slc = resp.into_body();
let reserve_resp: GhacReserveResponse = serde_json::from_reader(slc.reader())
@@ -286,7 +286,7 @@ impl GhacCore {
.extension(ServiceOperation("CreateCacheEntry"))
.body(body)
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let location = if resp.status() == StatusCode::OK {
let (parts, slc) = resp.into_parts();
let query_resp = ghac_types::CreateCacheEntryResponse::decode(slc)
@@ -332,7 +332,7 @@ impl GhacCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn ghac_finalize_upload(
@@ -358,7 +358,7 @@ impl GhacCore {
.extension(ServiceOperation("CommitCache"))
.body(Buffer::from(bs))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
if resp.status().is_success() {
Ok(())
} else {
@@ -388,7 +388,7 @@ impl GhacCore {
.extension(ServiceOperation("FinalizeCacheEntryUpload"))
.body(body)
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
if resp.status() != StatusCode::OK {
return Err(parse_error(resp));
};
diff --git a/core/services/github/src/core.rs b/core/services/github/src/core.rs
index 04ed26fc9..0ca2e252a 100644
--- a/core/services/github/src/core.rs
+++ b/core/services/github/src/core.rs
@@ -64,7 +64,7 @@ impl GithubCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub fn sign(&self, req: request::Builder) -> Result<request::Builder> {
@@ -164,7 +164,7 @@ impl GithubCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn upload(
diff --git a/core/services/hf/Cargo.toml b/core/services/hf/Cargo.toml
index b0d45cf12..c159da107 100644
--- a/core/services/hf/Cargo.toml
+++ b/core/services/hf/Cargo.toml
@@ -37,17 +37,14 @@ http = { workspace = true }
log = { workspace = true }
opendal-core = { path = "../../core", version = "0.57.0", default-features = false }
percent-encoding = "2"
-reqwest = { version = "0.13.4", default-features = false, features = [
- "rustls",
-] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
[dev-dependencies]
base64 = { workspace = true }
futures = { workspace = true }
-opendal-core = { path = "../../core", version = "0.57.0", features = [
- "reqwest-rustls-tls",
+opendal-http-transport-reqwest = { path = "../../http-transports/reqwest", version = "0.57.0", features = [
+ "rustls",
] }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
diff --git a/core/services/hf/src/backend.rs b/core/services/hf/src/backend.rs
index 16ab0c799..808230a7f 100644
--- a/core/services/hf/src/backend.rs
+++ b/core/services/hf/src/backend.rs
@@ -415,12 +415,14 @@ pub(super) mod test_utils {
use super::super::uri::{HfRepo, HfRepoType};
use super::HfBuilder;
use opendal_core::Capability;
+ use opendal_core::HttpTransporter;
use opendal_core::Operator;
- use opendal_core::raw::{HttpClient, ServiceInfo};
+ use opendal_core::raw::ServiceInfo;
fn finish_operator(op: Operator) -> Operator {
- let client = HttpClient::with(reqwest::Client::new());
- op.http_client(client)
+ let transport =
+ HttpTransporter::new(opendal_http_transport_reqwest::ReqwestTransport::default());
+ op.http_transport(transport)
}
pub fn gpt2_operator() -> Operator {
diff --git a/core/services/hf/src/core.rs b/core/services/hf/src/core.rs
index fb3f57384..c6b3c57eb 100644
--- a/core/services/hf/src/core.rs
+++ b/core/services/hf/src/core.rs
@@ -420,7 +420,7 @@ impl HfCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<HttpBody>> {
- let resp = ctx.http_client().fetch(req).await?;
+ let resp = ctx.http_transport().fetch(req).await?;
let (parts, body) = resp.into_parts();
if parts.status.is_success() {
Ok(Response::from_parts(parts, body))
@@ -492,7 +492,7 @@ impl HfCore {
}
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let resp = ctx.http_client().fetch(req).await?;
+ let resp = ctx.http_transport().fetch(req).await?;
if !resp.status().is_success() {
let status = resp.status();
@@ -589,12 +589,12 @@ pub(crate) mod test_utils {
use super::*;
#[derive(Clone)]
- pub(crate) struct MockHttpClient {
+ pub(crate) struct MockHttpTransport {
url: Arc<Mutex<Option<String>>>,
body: Arc<Mutex<Option<String>>>,
}
- impl MockHttpClient {
+ impl MockHttpTransport {
pub(crate) fn new() -> Self {
Self {
url: Arc::new(Mutex::new(None)),
@@ -611,7 +611,7 @@ pub(crate) mod test_utils {
}
}
- impl HttpFetch for MockHttpClient {
+ impl HttpTransport for MockHttpTransport {
async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
*self.url.lock().unwrap() = Some(req.uri().to_string());
*self.body.lock().unwrap() = Some(
@@ -662,10 +662,10 @@ pub(crate) mod test_utils {
repo_id: &str,
revision: &str,
endpoint: &str,
- ) -> (HfCore, OperationContext, MockHttpClient) {
- let mock_client = MockHttpClient::new();
- let http_client = HttpClient::with(mock_client.clone());
- let ctx = OperationContext::new(http_client, Executor::default());
+ ) -> (HfCore, OperationContext, MockHttpTransport) {
+ let mock_client = MockHttpTransport::new();
+ let http_transport = HttpTransporter::new(mock_client.clone());
+ let ctx = OperationContext::from_parts(http_transport, Executor::default());
let info = ServiceInfo::new("hf", "", "");
let capability = Capability::default();
diff --git a/core/services/hf/src/reader.rs b/core/services/hf/src/reader.rs
index aef49439a..e51f13cd3 100644
--- a/core/services/hf/src/reader.rs
+++ b/core/services/hf/src/reader.rs
@@ -169,10 +169,9 @@ mod tests {
use base64::Engine;
let core = testing_dataset_core();
- let ctx = OperationContext::new(
- HttpClient::with(reqwest::Client::new()),
- Executor::default(),
- );
+ let ctx = OperationContext::new().with_http_transport(HttpTransporter::new(
+ opendal_http_transport_reqwest::ReqwestTransport::default(),
+ ));
let content = b"non-xet fallback test content";
let path = "tests/non-xet-fallback.txt";
diff --git a/core/services/http/src/core.rs b/core/services/http/src/core.rs
index 6d6920ae5..02a07d58a 100644
--- a/core/services/http/src/core.rs
+++ b/core/services/http/src/core.rs
@@ -93,7 +93,7 @@ impl HttpCore {
args: &OpRead,
) -> Result<Response<HttpBody>> {
let req = self.http_get_request(path, range, args)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub fn http_head_request(&self, path: &str, args: &OpStat) -> Result<Request<Buffer>> {
@@ -129,6 +129,6 @@ impl HttpCore {
args: &OpStat,
) -> Result<Response<Buffer>> {
let req = self.http_head_request(path, args)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
diff --git a/core/services/ipfs/src/core.rs b/core/services/ipfs/src/core.rs
index 022d2c0e8..3599debf6 100644
--- a/core/services/ipfs/src/core.rs
+++ b/core/services/ipfs/src/core.rs
@@ -64,7 +64,7 @@ impl IpfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn ipfs_head(&self, ctx: &OperationContext, path: &str) -> Result<Response<Buffer>> {
@@ -78,7 +78,7 @@ impl IpfsCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn ipfs_list(&self, ctx: &OperationContext, path: &str) -> Result<Response<Buffer>> {
@@ -100,7 +100,7 @@ impl IpfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// IPFS's stat behavior highly depends on its implementation.
diff --git a/core/services/ipmfs/src/core.rs b/core/services/ipmfs/src/core.rs
index bb26494ca..5a2b6754b 100644
--- a/core/services/ipmfs/src/core.rs
+++ b/core/services/ipmfs/src/core.rs
@@ -55,7 +55,7 @@ impl IpmfsCore {
.extension(ServiceOperation("Stat"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn ipmfs_read(
@@ -82,7 +82,7 @@ impl IpmfsCore {
.extension(ServiceOperation("Read"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn ipmfs_rm(&self, ctx: &OperationContext, path: &str) -> Result<Response<Buffer>> {
@@ -99,7 +99,7 @@ impl IpmfsCore {
.extension(ServiceOperation("Rm"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub(crate) async fn ipmfs_ls(
@@ -120,7 +120,7 @@ impl IpmfsCore {
.extension(ServiceOperation("Ls"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn ipmfs_mkdir(
@@ -141,7 +141,7 @@ impl IpmfsCore {
.extension(ServiceOperation("Mkdir"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Support write from reader.
@@ -166,6 +166,6 @@ impl IpmfsCore {
.extension(ServiceOperation("Write"));
let req = multipart.apply(req)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
diff --git a/core/services/koofr/src/core.rs b/core/services/koofr/src/core.rs
index 7281156c4..c3dd33862 100644
--- a/core/services/koofr/src/core.rs
+++ b/core/services/koofr/src/core.rs
@@ -72,7 +72,7 @@ impl KoofrCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn get_mount_id(&self, ctx: &OperationContext) -> Result<&String> {
@@ -135,7 +135,7 @@ impl KoofrCore {
.body(Buffer::from(Bytes::from(bs)))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(auth_req).await?;
+ let resp = ctx.http_transport().send(auth_req).await?;
let status = resp.status();
@@ -212,7 +212,7 @@ impl KoofrCore {
.body(Buffer::from(Bytes::from(bs)))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -278,7 +278,7 @@ impl KoofrCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn put(
diff --git a/core/services/lakefs/src/core.rs b/core/services/lakefs/src/core.rs
index 655685f49..1a48f839a 100644
--- a/core/services/lakefs/src/core.rs
+++ b/core/services/lakefs/src/core.rs
@@ -77,7 +77,7 @@ impl LakefsCore {
.extension(ServiceOperation("StatObject"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn get_object_content(
@@ -113,7 +113,7 @@ impl LakefsCore {
.extension(ServiceOperation("GetObject"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn list_objects(
@@ -157,7 +157,7 @@ impl LakefsCore {
.extension(ServiceOperation("ListObjects"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn upload_object(
@@ -189,7 +189,7 @@ impl LakefsCore {
.extension(ServiceOperation("UploadObject"));
let req = req.body(body).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn delete_object(
@@ -220,7 +220,7 @@ impl LakefsCore {
.extension(ServiceOperation("DeleteObject"));
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn copy_object(
@@ -258,7 +258,7 @@ impl LakefsCore {
.extension(ServiceOperation("CopyObject"))
.body(serde_json::to_vec(&map).unwrap().into())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Parse LakefsStatus into Metadata
diff --git a/core/services/obs/src/core.rs b/core/services/obs/src/core.rs
index c09281880..e02086fd5 100644
--- a/core/services/obs/src/core.rs
+++ b/core/services/obs/src/core.rs
@@ -63,7 +63,7 @@ impl ObsCore {
self.signer.clone().with_context(
Context::new()
.with_file_read(reqsign_file_read_tokio::TokioFileRead)
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone()))
+ .with_http_send(ctx.http_transport().clone())
.with_env(reqsign_core::OsEnv),
)
}
@@ -101,7 +101,7 @@ impl ObsCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
@@ -117,7 +117,7 @@ impl ObsCore {
let req = self.sign(ctx, req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub fn obs_get_object_request(
diff --git a/core/services/onedrive/src/core.rs b/core/services/onedrive/src/core.rs
index 76d0926a9..2fb5a64d8 100644
--- a/core/services/onedrive/src/core.rs
+++ b/core/services/onedrive/src/core.rs
@@ -105,7 +105,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
/// Create a directory at path if not exist, return the metadata about the folder
@@ -197,7 +197,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- let response = ctx.http_client().send(request).await?;
+ let response = ctx.http_transport().send(request).await?;
if !response.status().is_success() {
return Err(parse_error(response));
}
@@ -263,7 +263,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- let response = ctx.http_client().send(request).await?;
+ let response = ctx.http_transport().send(request).await?;
let decoded_response: GraphApiOneDriveVersionsResponse =
serde_json::from_reader(response.into_body().reader())
.map_err(new_json_deserialize_error)?;
@@ -283,7 +283,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
/// Download a file
@@ -318,7 +318,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().fetch(request).await
+ ctx.http_transport().fetch(request).await
}
/// Upload a file
@@ -371,7 +371,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
#[allow(clippy::too_many_arguments)]
@@ -404,7 +404,7 @@ impl OneDriveCore {
.map_err(new_request_build_error)?;
// OneDrive documentation requires not sending the `Authorization` header
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
/// Create a upload session for chunk uploads
@@ -441,7 +441,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
/// Create a directory
@@ -478,7 +478,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
/// Delete a `DriveItem`
@@ -499,7 +499,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- ctx.http_client().send(request).await
+ ctx.http_transport().send(request).await
}
/// Initialize a copy
@@ -565,7 +565,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- let response = ctx.http_client().send(request).await?;
+ let response = ctx.http_transport().send(request).await?;
match response.status() {
StatusCode::ACCEPTED => parse_location(response.headers())?
.ok_or_else(|| {
@@ -594,7 +594,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- let response = ctx.http_client().send(request).await?;
+ let response = ctx.http_transport().send(request).await?;
let status: OneDriveMonitorStatus =
serde_json::from_reader(response.into_body().reader())
.map_err(new_json_deserialize_error)?;
@@ -653,7 +653,7 @@ impl OneDriveCore {
self.sign(ctx, &mut request).await?;
- let response = ctx.http_client().send(request).await?;
+ let response = ctx.http_transport().send(request).await?;
match response.status() {
// can get etag, metadata, etc...
StatusCode::OK => Ok(()),
@@ -705,7 +705,7 @@ impl OneDriveSigner {
.body(Buffer::from(encoded_payload))
.map_err(new_request_build_error)?;
- let response = ctx.http_client().send(request).await?;
+ let response = ctx.http_transport().send(request).await?;
match response.status() {
StatusCode::OK => {
let resp_body = response.into_body();
diff --git a/core/services/oss/src/core.rs b/core/services/oss/src/core.rs
index 8a95c9dd2..52a36a7b0 100644
--- a/core/services/oss/src/core.rs
+++ b/core/services/oss/src/core.rs
@@ -96,7 +96,7 @@ impl OssCore {
self.signer.clone().with_context(
self.sign_ctx
.clone()
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone())),
+ .with_http_send(ctx.http_transport().clone()),
)
}
@@ -141,7 +141,7 @@ impl OssCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Set sse headers
@@ -501,7 +501,7 @@ impl OssCore {
) -> Result<Response<HttpBody>> {
let req = self.oss_get_object_request(path, false, range, args)?;
let req = self.sign(ctx, req).await?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn oss_head_object(
diff --git a/core/services/pcloud/src/core.rs b/core/services/pcloud/src/core.rs
index 6ebba5560..c23d4a80d 100644
--- a/core/services/pcloud/src/core.rs
+++ b/core/services/pcloud/src/core.rs
@@ -61,7 +61,7 @@ impl PcloudCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
@@ -131,7 +131,7 @@ impl PcloudCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn ensure_dir_exists(&self, ctx: &OperationContext, path: &str) -> Result<()> {
diff --git a/core/services/s3/src/backend.rs b/core/services/s3/src/backend.rs
index 2e4f2fa30..8d0c29eb6 100644
--- a/core/services/s3/src/backend.rs
+++ b/core/services/s3/src/backend.rs
@@ -683,7 +683,7 @@ impl S3Builder {
// Try to detect region by HeadBucket.
let req = http::Request::head(&url).body(Buffer::new()).ok()?;
- let client = HttpClient::new().ok()?;
+ let client = HttpTransporter::default();
let res = client
.send(req)
.await
@@ -876,11 +876,7 @@ impl Builder for S3Builder {
// The assume-role provider owns its STS signer, so give it a
// concrete HTTP sender instead of relying on a future operation
// context.
- let sts_ctx = ctx
- .clone()
- .with_http_send(HttpClientHttpSend::new(HttpClient::with(
- GLOBAL_REQWEST_CLIENT.clone(),
- )));
+ let sts_ctx = ctx.clone().with_http_send(HttpTransporter::default());
let sts_request_signer = AwsV4Signer::new("sts", ®ion);
let sts_signer = Signer::new(sts_ctx, provider, sts_request_signer);
let mut assume_role_provider =
@@ -1401,7 +1397,7 @@ mod tests {
let op = OpWrite::default().with_content_type("application/json");
let args = OpPresign::new(op, Duration::from_secs(3600));
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
let presigned = backend
.presign(&ctx, "test.txt", args)
.await
diff --git a/core/services/s3/src/core.rs b/core/services/s3/src/core.rs
index 3374d43dd..d7ca65b81 100644
--- a/core/services/s3/src/core.rs
+++ b/core/services/s3/src/core.rs
@@ -140,7 +140,7 @@ impl S3Core {
self.signer.clone().with_context(
Context::new()
.with_file_read(reqsign_file_read_tokio::TokioFileRead)
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone()))
+ .with_http_send(ctx.http_transport().clone())
.with_env(reqsign_core::OsEnv),
)
}
@@ -180,7 +180,7 @@ impl S3Core {
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
if self.skip_signature {
- return ctx.http_client().send(req).await;
+ return ctx.http_transport().send(req).await;
}
let (mut parts, body) = req.into_parts();
@@ -198,7 +198,7 @@ impl S3Core {
// contains host header.
parts.headers.remove(HOST);
- ctx.http_client()
+ ctx.http_transport()
.send(Request::from_parts(parts, body))
.await
}
@@ -209,7 +209,7 @@ impl S3Core {
req: Request<Buffer>,
) -> Result<Response<HttpBody>> {
if self.skip_signature {
- return ctx.http_client().fetch(req).await;
+ return ctx.http_transport().fetch(req).await;
}
let (mut parts, body) = req.into_parts();
@@ -227,7 +227,7 @@ impl S3Core {
// contains host header.
parts.headers.remove(HOST);
- ctx.http_client()
+ ctx.http_transport()
.fetch(Request::from_parts(parts, body))
.await
}
diff --git a/core/services/seafile/src/core.rs b/core/services/seafile/src/core.rs
index bc40da0af..b496fac45 100644
--- a/core/services/seafile/src/core.rs
+++ b/core/services/seafile/src/core.rs
@@ -69,7 +69,7 @@ impl SeafileCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// get auth info
@@ -95,7 +95,7 @@ impl SeafileCore {
.body(Buffer::from(Bytes::from(body)))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
@@ -124,7 +124,7 @@ impl SeafileCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -313,7 +313,7 @@ impl SeafileCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
/// file detail
diff --git a/core/services/sqlite/src/backend.rs b/core/services/sqlite/src/backend.rs
index 8c55c7a7a..bbd51a18a 100644
--- a/core/services/sqlite/src/backend.rs
+++ b/core/services/sqlite/src/backend.rs
@@ -492,7 +492,7 @@ mod test {
.unwrap();
let accessor = SqliteBackend::new(core);
- let ctx = OperationContext::new(HttpClient::default(), Executor::default());
+ let ctx = OperationContext::new();
let mut writer = accessor.write(&ctx, "hello", OpWrite::default()).unwrap();
writer.write(Buffer::from("hello world")).await.unwrap();
writer.close().await.unwrap();
diff --git a/core/services/swift/src/core.rs b/core/services/swift/src/core.rs
index 4319dbc4d..4969b4193 100644
--- a/core/services/swift/src/core.rs
+++ b/core/services/swift/src/core.rs
@@ -151,7 +151,7 @@ impl SwiftCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Bulk delete multiple objects in a single request.
@@ -190,7 +190,7 @@ impl SwiftCore {
.body(Buffer::from(bytes::Bytes::from(body_str)))
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn swift_list(
@@ -227,7 +227,7 @@ impl SwiftCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn swift_create_object(
@@ -277,7 +277,7 @@ impl SwiftCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn swift_read(
@@ -325,7 +325,7 @@ impl SwiftCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn swift_copy(
@@ -371,7 +371,7 @@ impl SwiftCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn swift_get_metadata(
@@ -412,7 +412,7 @@ impl SwiftCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Build the segment path for an SLO part.
@@ -458,7 +458,7 @@ impl SwiftCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Finalize an SLO by uploading the manifest.
@@ -502,7 +502,7 @@ impl SwiftCore {
.body(Buffer::from(bytes::Bytes::from(body)))
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Delete an SLO manifest and all its segments.
@@ -538,7 +538,7 @@ impl SwiftCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
if !resp.status().is_success() {
return Ok(());
}
@@ -566,7 +566,7 @@ impl SwiftCore {
.map_err(new_request_build_error)?;
// Best effort — ignore individual segment delete failures.
- let _ = ctx.http_client().send(req).await;
+ let _ = ctx.http_transport().send(req).await;
}
}
diff --git a/core/services/tos/src/core.rs b/core/services/tos/src/core.rs
index cb80992a2..3995c511c 100644
--- a/core/services/tos/src/core.rs
+++ b/core/services/tos/src/core.rs
@@ -94,7 +94,7 @@ impl TosCore {
self.signer.clone().with_context(
Context::new()
.with_file_read(reqsign_file_read_tokio::TokioFileRead)
- .with_http_send(HttpClientHttpSend::new(ctx.http_client().clone()))
+ .with_http_send(ctx.http_transport().clone())
.with_env(OsEnv),
)
}
@@ -120,7 +120,7 @@ impl TosCore {
);
let resp = ctx
- .http_client()
+ .http_transport()
.send(Request::from_parts(parts, body))
.await?;
@@ -149,7 +149,7 @@ impl TosCore {
.expect("user agent must be valid header value"),
);
- ctx.http_client()
+ ctx.http_transport()
.fetch(Request::from_parts(parts, body))
.await
}
diff --git a/core/services/upyun/src/core.rs b/core/services/upyun/src/core.rs
index 72ef90053..2015afee4 100644
--- a/core/services/upyun/src/core.rs
+++ b/core/services/upyun/src/core.rs
@@ -85,7 +85,7 @@ impl UpyunCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub fn sign(&self, req: &mut Request<Buffer>) -> Result<()> {
@@ -129,7 +129,7 @@ impl UpyunCore {
self.sign(&mut req)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn info(&self, ctx: &OperationContext, path: &str) -> Result<Response<Buffer>> {
diff --git a/core/services/vercel-artifacts/src/core.rs b/core/services/vercel-artifacts/src/core.rs
index 1bf894505..2c65f5da2 100644
--- a/core/services/vercel-artifacts/src/core.rs
+++ b/core/services/vercel-artifacts/src/core.rs
@@ -69,7 +69,7 @@ impl VercelArtifactsCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub(crate) async fn vercel_artifacts_put(
@@ -99,7 +99,7 @@ impl VercelArtifactsCore {
let req = req.body(body).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub(crate) async fn vercel_artifacts_stat(
@@ -126,6 +126,6 @@ impl VercelArtifactsCore {
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
diff --git a/core/services/vercel-blob/src/core.rs b/core/services/vercel-blob/src/core.rs
index aeeb8a347..0dd252706 100644
--- a/core/services/vercel-blob/src/core.rs
+++ b/core/services/vercel-blob/src/core.rs
@@ -78,7 +78,7 @@ impl VercelBlobCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub fn sign(&self, req: request::Builder) -> request::Builder {
@@ -119,7 +119,7 @@ impl VercelBlobCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn upload(
diff --git a/core/services/webdav/src/core.rs b/core/services/webdav/src/core.rs
index f7037979b..327a77545 100644
--- a/core/services/webdav/src/core.rs
+++ b/core/services/webdav/src/core.rs
@@ -139,7 +139,7 @@ impl WebdavCore {
.body(Buffer::from(Bytes::from(PROPFIND_REQUEST)))
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
if !resp.status().is_success() {
return Err(parse_error(resp));
}
@@ -193,7 +193,7 @@ impl WebdavCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn webdav_put(
@@ -231,7 +231,7 @@ impl WebdavCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Set user-defined metadata using WebDAV PROPPATCH method.
@@ -271,7 +271,7 @@ impl WebdavCore {
.body(Buffer::from(Bytes::from(proppatch_body)))
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webdav_delete(
@@ -294,7 +294,7 @@ impl WebdavCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webdav_copy(
@@ -329,7 +329,7 @@ impl WebdavCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webdav_move(
@@ -364,7 +364,7 @@ impl WebdavCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webdav_list(
@@ -396,7 +396,7 @@ impl WebdavCore {
.body(Buffer::from(Bytes::from(PROPFIND_REQUEST)))
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// Create dir recursively for given path.
@@ -460,7 +460,7 @@ impl WebdavCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
match status {
diff --git a/core/services/webhdfs/src/core.rs b/core/services/webhdfs/src/core.rs
index 6f3dca2db..75a0493b7 100644
--- a/core/services/webhdfs/src/core.rs
+++ b/core/services/webhdfs/src/core.rs
@@ -80,7 +80,7 @@ impl WebhdfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// create object
@@ -114,7 +114,7 @@ impl WebhdfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -142,7 +142,7 @@ impl WebhdfsCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webhdfs_rename_object(
@@ -171,7 +171,7 @@ impl WebhdfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
async fn webhdfs_init_append(&self, ctx: &OperationContext, path: &str) -> Result<String> {
@@ -194,7 +194,7 @@ impl WebhdfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_client().send(req).await?;
+ let resp = ctx.http_transport().send(req).await?;
let status = resp.status();
@@ -235,7 +235,7 @@ impl WebhdfsCore {
.body(body)
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
/// CONCAT will concat sources to the path
@@ -274,7 +274,7 @@ impl WebhdfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webhdfs_list_status(
@@ -298,7 +298,7 @@ impl WebhdfsCore {
let req = Request::get(&url)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webhdfs_list_status_batch(
@@ -327,7 +327,7 @@ impl WebhdfsCore {
let req = Request::get(&url)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
fn webhdfs_open_request(&self, path: &str, range: &BytesRange) -> Result<Request<Buffer>> {
@@ -367,7 +367,7 @@ impl WebhdfsCore {
range: BytesRange,
) -> Result<Response<HttpBody>> {
let req = self.webhdfs_open_request(path, &range)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub(super) async fn webhdfs_get_file_status(
@@ -394,7 +394,7 @@ impl WebhdfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
pub async fn webhdfs_delete(
@@ -421,7 +421,7 @@ impl WebhdfsCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
}
diff --git a/core/services/yandex-disk/src/core.rs b/core/services/yandex-disk/src/core.rs
index 68fcdaa85..431d26532 100644
--- a/core/services/yandex-disk/src/core.rs
+++ b/core/services/yandex-disk/src/core.rs
@@ -54,7 +54,7 @@ impl YandexDiskCore {
ctx: &OperationContext,
req: Request<Buffer>,
) -> Result<Response<Buffer>> {
- ctx.http_client().send(req).await
+ ctx.http_transport().send(req).await
}
#[inline]
@@ -170,7 +170,7 @@ impl YandexDiskCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- ctx.http_client().fetch(req).await
+ ctx.http_transport().fetch(req).await
}
pub async fn ensure_dir_exists(&self, ctx: &OperationContext, path: &str) -> Result<()> {
diff --git a/core/src/lib.rs b/core/src/lib.rs
index 0fef6bbbf..2457fd2e1 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -27,6 +27,18 @@ pub use opendal_core::*;
#[cfg(feature = "tests")]
pub extern crate opendal_testkit as tests;
+/// Install the global defaults provided by the facade crate.
+///
+/// This function is safe to call multiple times. It registers enabled services
+/// and installs the default HTTP transport when the corresponding feature is
+/// enabled.
+pub fn install_default() {
+ init_default_registry();
+
+ #[cfg(feature = "http-transport-reqwest")]
+ HttpTransporter::install_default(opendal_http_transport_reqwest::ReqwestTransport::default());
+}
+
/// Initialize the global [`OperatorRegistry`] with enabled services.
///
/// This function is safe to call multiple times and will only perform
@@ -242,7 +254,7 @@ fn init_default_registry_inner(registry: &OperatorRegistry) {
#[cfg(feature = "auto-register-services")]
#[ctor::ctor(unsafe)]
fn register_default_operator_registry() {
- init_default_registry();
+ install_default();
}
/// Re-export of service implementations.