This is an automated email from the ASF dual-hosted git repository. koushiro pushed a commit to branch extract-logging-retry-timeout-layers in repository https://gitbox.apache.org/repos/asf/opendal.git
commit cc8e929e5866f86d076ff2f58677c0afdc69e32c Author: koushiro <[email protected]> AuthorDate: Wed Dec 17 17:38:13 2025 +0800 refactor: Split logging/retry/timeout layer to new crates --- core/Cargo.lock | 36 ++++++++++++++ core/Cargo.toml | 6 +++ core/core/Cargo.toml | 1 - core/core/src/layers/mod.rs | 11 ----- core/core/src/lib.rs | 20 +++++--- core/core/src/raw/ops.rs | 2 +- core/core/src/types/execute/executor.rs | 2 +- core/core/src/types/operator/builder.rs | 37 ++++++++------ core/core/src/types/operator/operator.rs | 19 ++++---- core/layers/logging/Cargo.toml | 21 ++++++++ .../logging.rs => layers/logging/src/lib.rs} | 19 ++++---- core/layers/retry/Cargo.toml | 28 +++++++++++ .../layers/retry.rs => layers/retry/src/lib.rs} | 39 +++++++-------- core/layers/timeout/Cargo.toml | 24 +++++++++ .../timeout.rs => layers/timeout/src/lib.rs} | 57 +++++++++++----------- core/src/lib.rs | 6 +++ 16 files changed, 220 insertions(+), 108 deletions(-) diff --git a/core/Cargo.lock b/core/Cargo.lock index 42dda365d..242e630ce 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -5541,6 +5541,7 @@ dependencies = [ "opendal-layer-fastmetrics", "opendal-layer-fastrace", "opendal-layer-immutable-index", + "opendal-layer-logging", "opendal-layer-metrics", "opendal-layer-mime-guess", "opendal-layer-observe-metrics-common", @@ -5548,8 +5549,10 @@ dependencies = [ "opendal-layer-oteltrace", "opendal-layer-prometheus", "opendal-layer-prometheus-client", + "opendal-layer-retry", "opendal-layer-tail-cut", "opendal-layer-throttle", + "opendal-layer-timeout", "opendal-layer-tracing", "opendal-service-aliyun-drive", "opendal-service-alluxio", @@ -5783,6 +5786,14 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "opendal-layer-logging" +version = "0.55.0" +dependencies = [ + "log", + "opendal-core", +] + [[package]] name = "opendal-layer-metrics" version = "0.55.0" @@ -5846,6 +5857,21 @@ dependencies = [ "prometheus-client", ] +[[package]] +name = "opendal-layer-retry" +version = "0.55.0" +dependencies = [ + "backon", + "bytes", + "futures", + "log", + "opendal-core", + "opendal-layer-logging", + "opendal-layer-timeout", + "tokio", + "tracing-subscriber", +] + [[package]] name = "opendal-layer-tail-cut" version = "0.55.0" @@ -5862,6 +5888,16 @@ dependencies = [ "opendal-core", ] +[[package]] +name = "opendal-layer-timeout" +version = "0.55.0" +dependencies = [ + "futures", + "opendal-core", + "opendal-layer-retry", + "tokio", +] + [[package]] name = "opendal-layer-tracing" version = "0.55.0" diff --git a/core/Cargo.toml b/core/Cargo.toml index d3bfbae2e..2b981cc15 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -91,14 +91,17 @@ layers-dtrace = ["opendal-core/layers-dtrace"] layers-fastmetrics = ["dep:opendal-layer-fastmetrics"] layers-fastrace = ["dep:opendal-layer-fastrace"] layers-immutable-index = ["dep:opendal-layer-immutable-index"] +layers-logging = ["dep:opendal-layer-logging"] layers-metrics = ["dep:opendal-layer-metrics"] layers-mime-guess = ["dep:opendal-layer-mime-guess"] layers-otel-metrics = ["dep:opendal-layer-otelmetrics"] layers-otel-trace = ["dep:opendal-layer-oteltrace"] layers-prometheus = ["dep:opendal-layer-prometheus"] layers-prometheus-client = ["dep:opendal-layer-prometheus-client"] +layers-retry = ["dep:opendal-layer-retry"] 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-tls = ["opendal-core/reqwest-rustls-tls"] services-aliyun-drive = ["dep:opendal-service-aliyun-drive"] @@ -193,6 +196,7 @@ opendal-layer-chaos = { path = "layers/chaos", version = "0.55.0", optional = tr opendal-layer-fastmetrics = { path = "layers/fastmetrics", version = "0.55.0", optional = true, default-features = false } opendal-layer-fastrace = { path = "layers/fastrace", version = "0.55.0", optional = true, default-features = false } opendal-layer-immutable-index = { path = "layers/immutable-index", version = "0.55.0", optional = true, default-features = false } +opendal-layer-logging = { path = "layers/logging", version = "0.55.0", optional = true, default-features = false } opendal-layer-metrics = { path = "layers/metrics", version = "0.55.0", optional = true, default-features = false } opendal-layer-mime-guess = { path = "layers/mime-guess", version = "0.55.0", optional = true, default-features = false } opendal-layer-observe-metrics-common = { path = "layers/observe-metrics-common", version = "0.55.0", optional = true, default-features = false } @@ -200,8 +204,10 @@ opendal-layer-otelmetrics = { path = "layers/otelmetrics", version = "0.55.0", o opendal-layer-oteltrace = { path = "layers/oteltrace", version = "0.55.0", optional = true, default-features = false } opendal-layer-prometheus = { path = "layers/prometheus", version = "0.55.0", optional = true, default-features = false } opendal-layer-prometheus-client = { path = "layers/prometheus-client", version = "0.55.0", optional = true, default-features = false } +opendal-layer-retry = { path = "layers/retry", version = "0.55.0", optional = true, default-features = false } opendal-layer-tail-cut = { path = "layers/tail-cut", version = "0.55.0", optional = true, default-features = false } opendal-layer-throttle = { path = "layers/throttle", version = "0.55.0", optional = true, default-features = false } +opendal-layer-timeout = { path = "layers/timeout", version = "0.55.0", optional = true, default-features = false } opendal-layer-tracing = { path = "layers/tracing", version = "0.55.0", optional = true, default-features = false } opendal-service-aliyun-drive = { path = "services/aliyun-drive", version = "0.55.0", optional = true, default-features = false } opendal-service-alluxio = { path = "services/alluxio", version = "0.55.0", optional = true, default-features = false } diff --git a/core/core/Cargo.toml b/core/core/Cargo.toml index 732cae0cc..0ffd7af65 100644 --- a/core/core/Cargo.toml +++ b/core/core/Cargo.toml @@ -89,7 +89,6 @@ doctest = false [dependencies] # Required dependencies anyhow = { version = "1.0.100", features = ["std"] } -backon = { version = "1.6", features = ["tokio-sleep"] } base64 = { workspace = true } bytes = { workspace = true } ctor = { workspace = true } diff --git a/core/core/src/layers/mod.rs b/core/core/src/layers/mod.rs index 368eacdaf..5d1f6e0df 100644 --- a/core/core/src/layers/mod.rs +++ b/core/core/src/layers/mod.rs @@ -32,17 +32,6 @@ pub use simulate::SimulateLayer; mod concurrent_limit; pub use concurrent_limit::ConcurrentLimitLayer; -mod logging; -pub use logging::LoggingInterceptor; -pub use logging::LoggingLayer; - -mod timeout; -pub use timeout::TimeoutLayer; - -mod retry; -pub use self::retry::RetryInterceptor; -pub use self::retry::RetryLayer; - #[cfg(all(target_os = "linux", feature = "layers-dtrace"))] mod dtrace; #[cfg(all(target_os = "linux", feature = "layers-dtrace"))] diff --git a/core/core/src/lib.rs b/core/core/src/lib.rs index fb7b08d05..5d9039efd 100644 --- a/core/core/src/lib.rs +++ b/core/core/src/lib.rs @@ -60,11 +60,12 @@ //! 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::LoggingLayer`] as an example; this layer adds logging to -//! every operation that OpenDAL performs. +//! Let's use [`layers::HttpClientLayer`] as an example; this layer allows +//! customizing the HTTP client used by OpenDAL. //! //! ```no_run -//! use opendal_core::layers::LoggingLayer; +//! use opendal_core::layers::HttpClientLayer; +//! use opendal_core::raw::HttpClient; //! use opendal_core::services; //! use opendal_core::Operator; //! use opendal_core::Result; @@ -75,9 +76,10 @@ //! let builder = services::Memory::default(); //! //! // Init an operator +//! let client = HttpClient::new()?; //! let op = Operator::new(builder)? -//! // Init with logging layer enabled. -//! .layer(LoggingLayer::default()) +//! // Init with custom HTTP client. +//! .layer(HttpClientLayer::new(client)) //! .finish(); //! //! Ok(()) @@ -102,8 +104,9 @@ //! into [`futures::AsyncRead`] or [`futures::Stream`] for broader ecosystem compatibility. //! //! ```no_run -//! use opendal_core::layers::LoggingLayer; +//! use opendal_core::layers::HttpClientLayer; //! use opendal_core::options; +//! use opendal_core::raw::HttpClient; //! use opendal_core::services; //! use opendal_core::Operator; //! use opendal_core::Result; @@ -114,9 +117,10 @@ //! let builder = services::Memory::default(); //! //! // Init an operator +//! let client = HttpClient::new()?; //! let op = Operator::new(builder)? -//! // Init with logging layer enabled. -//! .layer(LoggingLayer::default()) +//! // Init with custom HTTP client. +//! .layer(HttpClientLayer::new(client)) //! .finish(); //! //! // Fetch this file's metadata diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs index ae76a58f1..41182df6f 100644 --- a/core/core/src/raw/ops.rs +++ b/core/core/src/raw/ops.rs @@ -334,7 +334,7 @@ impl OpRead { } /// Returns a mutable range to allow updating. - pub(crate) fn range_mut(&mut self) -> &mut BytesRange { + pub fn range_mut(&mut self) -> &mut BytesRange { &mut self.range } diff --git a/core/core/src/types/execute/executor.rs b/core/core/src/types/execute/executor.rs index 1a1b69114..36a87f33e 100644 --- a/core/core/src/types/execute/executor.rs +++ b/core/core/src/types/execute/executor.rs @@ -74,7 +74,7 @@ impl Executor { } /// Return the inner executor. - pub(crate) fn into_inner(self) -> Arc<dyn Execute> { + pub fn into_inner(self) -> Arc<dyn Execute> { self.executor } diff --git a/core/core/src/types/operator/builder.rs b/core/core/src/types/operator/builder.rs index e90253d06..195006a8b 100644 --- a/core/core/src/types/operator/builder.rs +++ b/core/core/src/types/operator/builder.rs @@ -196,14 +196,17 @@ impl Operator { /// /// ```no_run /// # use std::sync::Arc; - /// # use anyhow::Result; - /// use opendal_core::layers::LoggingLayer; - /// use opendal_core::services::Memory; - /// use opendal_core::Operator; - /// + /// # + /// # use opendal_core::Result; + /// # use opendal_core::layers::HttpClientLayer; + /// # use opendal_core::raw::HttpClient; + /// # use opendal_core::services::Memory; + /// # use opendal_core::Operator; + /// # /// # async fn test() -> Result<()> { + /// let client = HttpClient::new()?; /// let op = Operator::new(Memory::default())?.finish(); - /// let op = op.layer(LoggingLayer::default()); + /// let op = op.layer(HttpClientLayer::new(client)); /// // All operations will go through the new_layer /// let _ = op.read("test_file").await?; /// # Ok(()) @@ -234,8 +237,6 @@ impl Operator { /// ``` /// use std::collections::HashMap; /// -/// use opendal_core::layers::LoggingLayer; -/// use opendal_core::layers::RetryLayer; /// use opendal_core::services; /// use opendal_core::Builder; /// use opendal_core::Operator; @@ -243,8 +244,9 @@ impl Operator { /// /// fn init_service<B: Builder>(cfg: HashMap<String, String>) -> Result<Operator> { /// let op = Operator::from_iter::<B>(cfg)? -/// .layer(LoggingLayer::default()) -/// .layer(RetryLayer::new()) +/// // add layers +/// // .layer(LoggingLayer::default()) +/// // .layer(RetryLayer::new()) /// .finish(); /// /// Ok(op) @@ -290,14 +292,17 @@ impl<A: Access> OperatorBuilder<A> { /// /// ```no_run /// # use std::sync::Arc; - /// # use anyhow::Result; - /// use opendal_core::layers::LoggingLayer; - /// use opendal_core::services::Memory; - /// use opendal_core::Operator; - /// + /// # + /// # use opendal_core::Result; + /// # use opendal_core::layers::HttpClientLayer; + /// # use opendal_core::raw::HttpClient; + /// # use opendal_core::services::Memory; + /// # use opendal_core::Operator; + /// # /// # async fn test() -> Result<()> { + /// let client = HttpClient::new()?; /// let op = Operator::new(Memory::default())? - /// .layer(LoggingLayer::default()) + /// .layer(HttpClientLayer::new(client)) /// .finish(); /// // All operations will go through the new_layer /// let _ = op.read("test_file").await?; diff --git a/core/core/src/types/operator/operator.rs b/core/core/src/types/operator/operator.rs index caf895e57..9891652ab 100644 --- a/core/core/src/types/operator/operator.rs +++ b/core/core/src/types/operator/operator.rs @@ -59,7 +59,7 @@ use crate::*; /// /// After the operator is built, users can add the layers they need on top of it. /// -/// OpenDAL offers various layers for users to choose from, such as `RetryLayer`, `LoggingLayer`, and more. Visit [`layers`] for further details. +/// OpenDAL offers various layers for users to choose from. Visit [`layers`] for further details. /// /// Please note that `Layer` can modify internal contexts such as `HttpClient` /// and `Runtime` for all clones of given operator. Therefore, it is recommended @@ -67,15 +67,18 @@ use crate::*; /// layers after accessing the storage may result in unexpected behavior. /// /// ``` -/// # use anyhow::Result; -/// use opendal_core::layers::RetryLayer; +/// use opendal_core::layers::HttpClientLayer; +/// use opendal_core::raw::HttpClient; /// use opendal_core::services::Memory; /// use opendal_core::Operator; +/// use opendal_core::Result; +/// /// async fn test() -> Result<()> { /// let op: Operator = Operator::new(Memory::default())?.finish(); /// -/// // OpenDAL will retry failed operations now. -/// let op = op.layer(RetryLayer::default()); +/// // OpenDAL will replace the default HTTP client now. +/// let client = HttpClient::new()?; +/// let op = op.layer(HttpClientLayer::new(client)); /// /// Ok(()) /// } @@ -101,7 +104,6 @@ use crate::*; /// into [`futures::AsyncRead`] or [`futures::Stream`] for broader ecosystem compatibility. /// /// ```no_run -/// use opendal_core::layers::LoggingLayer; /// use opendal_core::options; /// use opendal_core::services; /// use opendal_core::Operator; @@ -113,10 +115,7 @@ use crate::*; /// let builder = services::Memory::default(); /// /// // Init an operator -/// let op = Operator::new(builder)? -/// // Init with logging layer enabled. -/// .layer(LoggingLayer::default()) -/// .finish(); +/// let op = Operator::new(builder)?.finish(); /// /// // Fetch this file's metadata /// let meta = op.stat("hello.txt").await?; diff --git a/core/layers/logging/Cargo.toml b/core/layers/logging/Cargo.toml new file mode 100644 index 000000000..de3af5239 --- /dev/null +++ b/core/layers/logging/Cargo.toml @@ -0,0 +1,21 @@ +[package] +description = "Apache OpenDAL logging layer" +name = "opendal-layer-logging" + +authors = { workspace = true } +edition = { workspace = true } +homepage = { workspace = true } +license = { workspace = true } +repository = { workspace = true } +rust-version = { workspace = true } +version = { workspace = true } + +[package.metadata.docs.rs] +all-features = true + +[dependencies] +log = { workspace = true } +opendal-core = { path = "../../core", version = "0.55.0", default-features = false } + +[dev-dependencies] +opendal-core = { path = "../../core", version = "0.55.0" } diff --git a/core/core/src/layers/logging.rs b/core/layers/logging/src/lib.rs similarity index 98% rename from core/core/src/layers/logging.rs rename to core/layers/logging/src/lib.rs index 98da44320..34e112fc1 100644 --- a/core/core/src/layers/logging.rs +++ b/core/layers/logging/src/lib.rs @@ -21,9 +21,8 @@ use std::sync::Arc; use log::Level; use log::log; - -use crate::raw::*; -use crate::*; +use opendal_core::raw::*; +use opendal_core::*; /// Add [log](https://docs.rs/log/) for every operation. /// @@ -41,16 +40,16 @@ use crate::*; /// # Examples /// /// ```no_run -/// # use opendal_core::layers::LoggingLayer; /// # use opendal_core::services; /// # use opendal_core::Operator; /// # use opendal_core::Result; -/// +/// # use opendal_layer_logging::LoggingLayer; +/// # /// # fn main() -> Result<()> { /// let _ = Operator::new(services::Memory::default())? /// .layer(LoggingLayer::default()) /// .finish(); -/// Ok(()) +/// # Ok(()) /// # } /// ``` /// @@ -75,14 +74,14 @@ use crate::*; /// You can implement your own logging interceptor to customize the logging behavior. /// /// ```no_run -/// # use opendal_core::layers::LoggingInterceptor; -/// # use opendal_core::layers::LoggingLayer; /// # use opendal_core::raw; /// # use opendal_core::services; /// # use opendal_core::Error; /// # use opendal_core::Operator; /// # use opendal_core::Result; -/// +/// # use opendal_layer_logging::LoggingInterceptor; +/// # use opendal_layer_logging::LoggingLayer; +/// # /// #[derive(Debug, Clone)] /// struct MyLoggingInterceptor; /// @@ -103,7 +102,7 @@ use crate::*; /// let _ = Operator::new(services::Memory::default())? /// .layer(LoggingLayer::new(MyLoggingInterceptor)) /// .finish(); -/// Ok(()) +/// # Ok(()) /// # } /// ``` #[derive(Debug)] diff --git a/core/layers/retry/Cargo.toml b/core/layers/retry/Cargo.toml new file mode 100644 index 000000000..e80d0e08e --- /dev/null +++ b/core/layers/retry/Cargo.toml @@ -0,0 +1,28 @@ +[package] +description = "Apache OpenDAL retry layer" +name = "opendal-layer-retry" + +authors = { workspace = true } +edition = { workspace = true } +homepage = { workspace = true } +license = { workspace = true } +repository = { workspace = true } +rust-version = { workspace = true } +version = { workspace = true } + +[package.metadata.docs.rs] +all-features = true + +[dependencies] +backon = { version = "1.6", features = ["tokio-sleep"] } +log = { workspace = true } +opendal-core = { path = "../../core", version = "0.55.0", default-features = false } + +[dev-dependencies] +bytes = { workspace = true } +futures = { workspace = true, default-features = true } +opendal-core = { path = "../../core", version = "0.55.0" } +opendal-layer-logging = { path = "../logging", version = "0.55.0", default-features = false } +opendal-layer-timeout = { path = "../timeout", version = "0.55.0", default-features = false } +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tracing-subscriber = { version = "0.3", features = ["env-filter", "tracing-log"] } diff --git a/core/core/src/layers/retry.rs b/core/layers/retry/src/lib.rs similarity index 98% rename from core/core/src/layers/retry.rs rename to core/layers/retry/src/lib.rs index 126b0f267..1ba9e5864 100644 --- a/core/core/src/layers/retry.rs +++ b/core/layers/retry/src/lib.rs @@ -22,9 +22,8 @@ use std::sync::Arc; use backon::ExponentialBuilder; use backon::Retryable; use log::warn; - -use crate::raw::*; -use crate::*; +use opendal_core::raw::*; +use opendal_core::*; /// Add retry for temporary failed operations. /// @@ -47,13 +46,13 @@ use crate::*; /// /// ```no_run /// # use std::time::Duration; -/// -/// # use opendal_core::layers::RetryLayer; -/// # use opendal_core::layers::TimeoutLayer; +/// # /// # use opendal_core::services; /// # use opendal_core::Operator; /// # use opendal_core::Result; -/// +/// # use opendal_layer_retry::RetryLayer; +/// # use opendal_layer_timeout::TimeoutLayer; +/// # /// # fn main() -> Result<()> { /// let op = Operator::new(services::Memory::default())? /// // This is fine, since timeout happen during retry. @@ -62,23 +61,23 @@ use crate::*; /// // This is wrong. Since timeout layer will drop future, leaving retry layer in a bad state. /// .layer(TimeoutLayer::new().with_io_timeout(Duration::from_nanos(1))) /// .finish(); -/// Ok(()) +/// # Ok(()) /// # } /// ``` /// /// # Examples /// /// ```no_run -/// # use opendal_core::layers::RetryLayer; /// # use opendal_core::services; /// # use opendal_core::Operator; /// # use opendal_core::Result; -/// +/// # use opendal_layer_retry::RetryLayer; +/// # /// # fn main() -> Result<()> { /// let _ = Operator::new(services::Memory::default())? /// .layer(RetryLayer::new()) /// .finish(); -/// Ok(()) +/// # Ok(()) /// # } /// ``` /// @@ -89,14 +88,14 @@ use crate::*; /// /// ```no_run /// # use std::time::Duration; -/// -/// # use opendal_core::layers::RetryInterceptor; -/// # use opendal_core::layers::RetryLayer; +/// # /// # use opendal_core::services; /// # use opendal_core::Error; /// # use opendal_core::Operator; /// # use opendal_core::Result; -/// +/// # use opendal_layer_retry::RetryInterceptor; +/// # use opendal_layer_retry::RetryLayer; +/// # /// struct MyRetryInterceptor; /// /// impl RetryInterceptor for MyRetryInterceptor { @@ -109,7 +108,7 @@ use crate::*; /// let _ = Operator::new(services::Memory::default())? /// .layer(RetryLayer::new().with_notify(MyRetryInterceptor)) /// .finish(); -/// Ok(()) +/// # Ok(()) /// # } /// ``` pub struct RetryLayer<I: RetryInterceptor = DefaultRetryInterceptor> { @@ -140,10 +139,9 @@ impl RetryLayer { /// # Examples /// /// ```no_run - /// use anyhow::Result; - /// use opendal_core::layers::RetryLayer; /// use opendal_core::services; /// use opendal_core::Operator; + /// use opendal_layer_retry::RetryLayer; /// /// let _ = Operator::new(services::Memory::default()) /// .expect("must init") @@ -158,9 +156,9 @@ impl<I: RetryInterceptor> RetryLayer<I> { /// Set the retry interceptor as new notify. /// /// ```no_run - /// use opendal_core::layers::RetryLayer; /// use opendal_core::services; /// use opendal_core::Operator; + /// use opendal_layer_retry::RetryLayer; /// /// fn notify(_err: &opendal_core::Error, _dur: std::time::Duration) {} /// @@ -614,16 +612,15 @@ impl<P: oio::Delete, I: RetryInterceptor> oio::Delete for RetryWrapper<P, I> { #[cfg(test)] mod tests { - use std::sync::Arc; use std::sync::Mutex; use bytes::Bytes; use futures::TryStreamExt; use futures::stream; + use opendal_layer_logging::LoggingLayer; use tracing_subscriber::filter::LevelFilter; use super::*; - use crate::layers::LoggingLayer; #[derive(Default, Clone)] struct MockBuilder { diff --git a/core/layers/timeout/Cargo.toml b/core/layers/timeout/Cargo.toml new file mode 100644 index 000000000..eb9cf4c85 --- /dev/null +++ b/core/layers/timeout/Cargo.toml @@ -0,0 +1,24 @@ +[package] +description = "Apache OpenDAL timeout layer" +name = "opendal-layer-timeout" + +authors = { workspace = true } +edition = { workspace = true } +homepage = { workspace = true } +license = { workspace = true } +repository = { workspace = true } +rust-version = { workspace = true } +version = { workspace = true } + +[package.metadata.docs.rs] +all-features = true + +[dependencies] +opendal-core = { path = "../../core", version = "0.55.0", default-features = false } +tokio = { workspace = true, features = ["time"] } + +[dev-dependencies] +futures = { workspace = true } +opendal-core = { path = "../../core", version = "0.55.0" } +opendal-layer-retry = { path = "../retry", version = "0.55.0", default-features = false } +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/core/core/src/layers/timeout.rs b/core/layers/timeout/src/lib.rs similarity index 93% rename from core/core/src/layers/timeout.rs rename to core/layers/timeout/src/lib.rs index 9a6fae346..dc6c881c3 100644 --- a/core/core/src/layers/timeout.rs +++ b/core/layers/timeout/src/lib.rs @@ -17,9 +17,10 @@ use std::future::Future; use std::sync::Arc; +use std::time::Duration; -use crate::raw::*; -use crate::*; +use opendal_core::raw::*; +use opendal_core::*; /// Add timeout for every operation to avoid slow or unexpected hang operations. /// @@ -52,13 +53,13 @@ use crate::*; /// /// ```no_run /// # use std::time::Duration; -/// -/// # use opendal_core::layers::RetryLayer; -/// # use opendal_core::layers::TimeoutLayer; +/// # /// # use opendal_core::services; /// # use opendal_core::Operator; /// # use opendal_core::Result; -/// +/// # use opendal_layer_retry::RetryLayer; +/// # use opendal_layer_timeout::TimeoutLayer; +/// # /// # fn main() -> Result<()> { /// let op = Operator::new(services::Memory::default())? /// // This is fine, since timeout happen during retry. @@ -67,7 +68,7 @@ use crate::*; /// // This is wrong. Since timeout layer will drop future, leaving retry layer in a bad state. /// .layer(TimeoutLayer::new().with_io_timeout(Duration::from_nanos(1))) /// .finish(); -/// Ok(()) +/// # Ok(()) /// # } /// ``` /// @@ -78,12 +79,12 @@ use crate::*; /// /// ```no_run /// # use std::time::Duration; -/// -/// # use opendal_core::layers::TimeoutLayer; +/// # /// # use opendal_core::services; /// # use opendal_core::Operator; /// # use opendal_core::Result; -/// +/// # use opendal_layer_timeout::TimeoutLayer; +/// # /// # fn main() -> Result<()> { /// let _ = Operator::new(services::Memory::default())? /// .layer( @@ -92,7 +93,7 @@ use crate::*; /// .with_io_timeout(Duration::from_secs(3)), /// ) /// .finish(); -/// Ok(()) +/// # Ok(()) /// # } /// ``` /// @@ -354,25 +355,23 @@ impl<R: oio::Delete> oio::Delete for TimeoutWrapper<R> { mod tests { use std::future::Future; use std::future::pending; - use std::sync::Arc; use futures::StreamExt; + use opendal_core::raw::*; + use opendal_core::*; use tokio::time::sleep; use tokio::time::timeout; - use crate::layers::TimeoutLayer; - use crate::layers::TypeEraseLayer; - use crate::raw::*; - use crate::*; + use super::*; #[derive(Debug, Clone, Default)] struct MockService; impl Access for MockService { - type Reader = MockReader; - type Writer = (); - type Lister = MockLister; - type Deleter = (); + type Reader = oio::Reader; + type Writer = oio::Writer; + type Lister = oio::Lister; + type Deleter = oio::Deleter; fn info(&self) -> Arc<AccessorInfo> { let am = AccessorInfo::default(); @@ -387,18 +386,18 @@ mod tests { /// This function will build a reader that always return pending. async fn read(&self, _: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> { - Ok((RpRead::new(), MockReader)) + Ok((RpRead::new(), Box::new(MockReader))) } /// This function will never return. async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> { sleep(Duration::from_secs(u64::MAX)).await; - Ok((RpDelete::default(), ())) + Ok((RpDelete::default(), Box::new(()))) } async fn list(&self, _: &str, _: OpList) -> Result<(RpList, Self::Lister)> { - Ok((RpList::default(), MockLister)) + Ok((RpList::default(), Box::new(MockLister))) } } @@ -422,8 +421,8 @@ mod tests { #[tokio::test] async fn test_operation_timeout() { - let acc = Arc::new(TypeEraseLayer.layer(MockService)) as Accessor; - let op = Operator::from_inner(acc) + let srv = MockService; + let op = Operator::from_inner(Arc::new(srv)) .layer(TimeoutLayer::new().with_timeout(Duration::from_secs(1))); let fut = async { @@ -441,8 +440,8 @@ mod tests { #[tokio::test] async fn test_io_timeout() { - let acc = Arc::new(TypeEraseLayer.layer(MockService)) as Accessor; - let op = Operator::from_inner(acc) + let srv = MockService; + let op = Operator::from_inner(Arc::new(srv)) .layer(TimeoutLayer::new().with_io_timeout(Duration::from_secs(1))); let reader = op.reader("test").await.unwrap(); @@ -456,8 +455,8 @@ mod tests { #[tokio::test] async fn test_list_timeout() { - let acc = Arc::new(TypeEraseLayer.layer(MockService)) as Accessor; - let op = Operator::from_inner(acc).layer( + let srv = MockService; + let op = Operator::from_inner(Arc::new(srv)).layer( TimeoutLayer::new() .with_timeout(Duration::from_secs(1)) .with_io_timeout(Duration::from_secs(1)), diff --git a/core/src/lib.rs b/core/src/lib.rs index 870c4887b..5e973c37a 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -127,6 +127,8 @@ pub mod layers { pub use opendal_layer_fastrace::*; #[cfg(feature = "layers-immutable-index")] pub use opendal_layer_immutable_index::*; + #[cfg(feature = "layers-logging")] + pub use opendal_layer_logging::*; #[cfg(feature = "layers-metrics")] pub use opendal_layer_metrics::*; #[cfg(feature = "layers-mime-guess")] @@ -139,10 +141,14 @@ pub mod layers { pub use opendal_layer_prometheus::*; #[cfg(feature = "layers-prometheus-client")] pub use opendal_layer_prometheus_client::*; + #[cfg(feature = "layers-retry")] + pub use opendal_layer_retry::*; #[cfg(feature = "layers-tail-cut")] pub use opendal_layer_tail_cut::*; #[cfg(feature = "layers-throttle")] pub use opendal_layer_throttle::*; + #[cfg(feature = "layers-timeout")] + pub use opendal_layer_timeout::*; #[cfg(feature = "layers-tracing")] pub use opendal_layer_tracing::*; }
