Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2985fa0a0e | ||
|
|
7a063fa75c | ||
|
|
ebb3c8ad6b | ||
|
|
076b0bb757 |
@@ -0,0 +1,23 @@
|
|||||||
|
# Round 8 — COMPLETE
|
||||||
|
|
||||||
|
## New Feature
|
||||||
|
### ResilientHttpClient (mytheclipse-http, resilience feature)
|
||||||
|
- File: `crates/mytheclipse-http/src/resilient_client.rs`
|
||||||
|
- `ResilientClientConfig { timeout, max_attempts, rate_per_sec, rate_burst, circuit_breaker }`
|
||||||
|
- `ResilientHttpClient::new(config)` builds `ServiceBuilder` pipeline
|
||||||
|
- `send(req)`, `get(url)`, `post(url, body)` — all run through `ServiceBuilder::run`
|
||||||
|
- Error type `RunError<Box<dyn std::error::Error + Send + Sync>>`
|
||||||
|
- Feature: `resilience = ["dep:reqwest", "dep:tokio", "dep:mytheclipse"]`
|
||||||
|
- mytheclipse dep now `features=["full"]` (was observability)
|
||||||
|
- 2 tests (config defaults + build)
|
||||||
|
|
||||||
|
## Modified
|
||||||
|
- http/Cargo.toml — resilience feature + mytheclipse full features
|
||||||
|
- http/lib.rs — module + re-export
|
||||||
|
- core/lib.rs — pub use RunError, ServiceConfig (needed by http crate)
|
||||||
|
- error.rs — RateLimit(String) variant (queue crate, round 6 carryover)
|
||||||
|
|
||||||
|
## Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
|
||||||
|
|
||||||
|
## Skill created: rust-workspace-abstractions (software-development)
|
||||||
|
Captures feature-gating, cross-crate deps, trait/async patterns, ownership patterns, error types, testing conventions for workspace abstraction authoring.
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
# Implementation Spec: Round 9 — COMPLETE
|
||||||
|
|
||||||
|
## New Feature
|
||||||
|
|
||||||
|
### RetryExt (mytheclipse-core, resiliency)
|
||||||
|
File: `crates/mytheclipse/src/retry_ext.rs`
|
||||||
|
- `RetryExt` trait — `.retry(config, predicate, self_fn)` extension pada Future<Output=Result<T,E>>
|
||||||
|
- Delegasi ke `crate::retry::retry`
|
||||||
|
- Non-Send Pin<Box<...>> return (single-threaded test OK)
|
||||||
|
- 1 test (retries_then_succeeds)
|
||||||
|
|
||||||
|
## Files
|
||||||
|
- new: retry_ext.rs
|
||||||
|
- core/lib.rs: +module +pub use RetryExt
|
||||||
|
|
||||||
|
Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
|
||||||
@@ -1,3 +1,17 @@
|
|||||||
|
# [1.12.0](https://github.com/asepharyana/mytheclipse/compare/v1.11.0...v1.12.0) (2026-08-29)
|
||||||
|
|
||||||
|
|
||||||
|
### Features
|
||||||
|
|
||||||
|
* round-9 abstractions — RetryExt ergonomic retry, ResilientHttpClient ([7a063fa](https://github.com/asepharyana/mytheclipse/commit/7a063fa75c6ca243d1e76a485e117a3b334b25e9))
|
||||||
|
|
||||||
|
# [1.11.0](https://github.com/asepharyana/mytheclipse/compare/v1.10.0...v1.11.0) (2026-08-29)
|
||||||
|
|
||||||
|
|
||||||
|
### Features
|
||||||
|
|
||||||
|
* round-8 abstractions — ResilientHttpClient, MiddlewarePipeline, BgJoiner ([076b0bb](https://github.com/asepharyana/mytheclipse/commit/076b0bb75789ecf7f19f1b4078a3260d33218fb6))
|
||||||
|
|
||||||
# [1.10.0](https://github.com/asepharyana/mytheclipse/compare/v1.9.0...v1.10.0) (2026-08-29)
|
# [1.10.0](https://github.com/asepharyana/mytheclipse/compare/v1.9.0...v1.10.0) (2026-08-29)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Generated
+10
-10
@@ -2827,7 +2827,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse"
|
name = "mytheclipse"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"num_cpus",
|
"num_cpus",
|
||||||
@@ -2841,7 +2841,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-cache"
|
name = "mytheclipse-cache"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"moka",
|
"moka",
|
||||||
@@ -2854,7 +2854,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-cli"
|
name = "mytheclipse-cli"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"clap",
|
"clap",
|
||||||
"tokio",
|
"tokio",
|
||||||
@@ -2863,7 +2863,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-config"
|
name = "mytheclipse-config"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"dotenvy",
|
"dotenvy",
|
||||||
"notify",
|
"notify",
|
||||||
@@ -2878,7 +2878,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-crypto"
|
name = "mytheclipse-crypto"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"aead",
|
"aead",
|
||||||
"aes-gcm",
|
"aes-gcm",
|
||||||
@@ -2900,7 +2900,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-event"
|
name = "mytheclipse-event"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-nats",
|
"async-nats",
|
||||||
"async-trait",
|
"async-trait",
|
||||||
@@ -2916,7 +2916,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-http"
|
name = "mytheclipse-http"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"axum",
|
"axum",
|
||||||
@@ -2932,7 +2932,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-queue"
|
name = "mytheclipse-queue"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-nats",
|
"async-nats",
|
||||||
"async-trait",
|
"async-trait",
|
||||||
@@ -2948,7 +2948,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-storage"
|
name = "mytheclipse-storage"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"aws-config",
|
"aws-config",
|
||||||
@@ -2964,7 +2964,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-tracing"
|
name = "mytheclipse-tracing"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"opentelemetry 0.25.0",
|
"opentelemetry 0.25.0",
|
||||||
"tokio",
|
"tokio",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-cache"
|
name = "mytheclipse-cache"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-cli"
|
name = "mytheclipse-cli"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-config"
|
name = "mytheclipse-config"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-crypto"
|
name = "mytheclipse-crypto"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-event"
|
name = "mytheclipse-event"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-http"
|
name = "mytheclipse-http"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
@@ -23,6 +23,8 @@ server-hyper = ["dep:hyper", "dep:tokio"]
|
|||||||
server-axum = ["dep:axum", "dep:hyper", "dep:tokio"]
|
server-axum = ["dep:axum", "dep:hyper", "dep:tokio"]
|
||||||
# Metrics HTTP endpoint serving Prometheus text format from a MetricsCollector.
|
# Metrics HTTP endpoint serving Prometheus text format from a MetricsCollector.
|
||||||
metrics-http = ["dep:axum", "dep:tower", "dep:tokio", "dep:mytheclipse"]
|
metrics-http = ["dep:axum", "dep:tower", "dep:tokio", "dep:mytheclipse"]
|
||||||
|
# Resilient HTTP client integrating retry + circuit breaker + timeout.
|
||||||
|
resilience = ["dep:reqwest", "dep:tokio", "dep:mytheclipse"]
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
tracing = "0.1"
|
tracing = "0.1"
|
||||||
@@ -34,7 +36,7 @@ axum = { version = "0.8", optional = true }
|
|||||||
tower = { version = "0.5", optional = true, default-features = false, features = ["util"] }
|
tower = { version = "0.5", optional = true, default-features = false, features = ["util"] }
|
||||||
serde = { version = "1", features = ["derive"] }
|
serde = { version = "1", features = ["derive"] }
|
||||||
serde_json = "1"
|
serde_json = "1"
|
||||||
mytheclipse = { version = "1.5", path = "../mytheclipse", optional = true, default-features = false, features = ["observability"] }
|
mytheclipse = { version = "1.5", path = "../mytheclipse", optional = true, default-features = false, features = ["full"] }
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
tokio = { version = "1.53", features = ["full"] }
|
tokio = { version = "1.53", features = ["full"] }
|
||||||
|
|||||||
@@ -9,6 +9,11 @@
|
|||||||
//! mytheclipse-http = { version = "0.2", features = ["client"] }
|
//! mytheclipse-http = { version = "0.2", features = ["client"] }
|
||||||
//! ```
|
//! ```
|
||||||
|
|
||||||
|
#[cfg(feature = "resilience")]
|
||||||
|
pub mod resilient_client;
|
||||||
|
#[cfg(feature = "resilience")]
|
||||||
|
pub use resilient_client::{ResilientHttpClient, ResilientClientConfig};
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
pub mod client;
|
pub mod client;
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,129 @@
|
|||||||
|
//! Resilient HTTP client with retry + circuit breaker + timeout (feature `resilience`).
|
||||||
|
//!
|
||||||
|
//! Wraps `reqwest::Client` with `mytheclipse::ServiceBuilder`, applying retry,
|
||||||
|
//! circuit-breaker, and timeout layers around every request.
|
||||||
|
|
||||||
|
use std::pin::Pin;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use reqwest::Client;
|
||||||
|
use reqwest::Method;
|
||||||
|
use reqwest::RequestBuilder;
|
||||||
|
use tracing::Instrument;
|
||||||
|
|
||||||
|
use mytheclipse::{CircuitBreaker, RunError, ServiceBuilder, ServiceConfig};
|
||||||
|
|
||||||
|
type HttpError = Box<dyn std::error::Error + Send + Sync>;
|
||||||
|
|
||||||
|
/// Configuration for [`ResilientHttpClient`].
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct ResilientClientConfig {
|
||||||
|
pub timeout: Duration,
|
||||||
|
pub max_attempts: u32,
|
||||||
|
pub rate_per_sec: f64,
|
||||||
|
pub rate_burst: u64,
|
||||||
|
pub circuit_breaker: Option<CircuitBreaker>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for ResilientClientConfig {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
timeout: Duration::from_secs(30),
|
||||||
|
max_attempts: 1,
|
||||||
|
rate_per_sec: 0.0,
|
||||||
|
rate_burst: 0,
|
||||||
|
circuit_breaker: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A reqwest client that runs every request through a `ServiceBuilder`
|
||||||
|
/// pipeline (retry + circuit breaker + timeout).
|
||||||
|
pub struct ResilientHttpClient {
|
||||||
|
inner: Client,
|
||||||
|
config: ResilientClientConfig,
|
||||||
|
builder: ServiceBuilder,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ResilientHttpClient {
|
||||||
|
/// Creates a new resilient client from the given config.
|
||||||
|
pub fn new(config: ResilientClientConfig) -> Self {
|
||||||
|
let svc_cfg = ServiceConfig {
|
||||||
|
max_attempts: config.max_attempts,
|
||||||
|
timeout: config.timeout,
|
||||||
|
rate_per_sec: config.rate_per_sec,
|
||||||
|
rate_burst: config.rate_burst,
|
||||||
|
};
|
||||||
|
let mut builder = ServiceBuilder::new(svc_cfg);
|
||||||
|
if let Some(cb) = &config.circuit_breaker {
|
||||||
|
builder = builder.with_circuit_breaker(cb.clone());
|
||||||
|
}
|
||||||
|
Self {
|
||||||
|
inner: Client::new(),
|
||||||
|
config,
|
||||||
|
builder,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the configured default timeout.
|
||||||
|
pub fn timeout(&self) -> Duration {
|
||||||
|
self.config.timeout
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns a `RequestBuilder` for `method` + `url`.
|
||||||
|
pub fn request(&self, method: Method, url: &str) -> RequestBuilder {
|
||||||
|
self.inner.request(method, url)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sends a pre-built `RequestBuilder` through the resiliency pipeline.
|
||||||
|
/// Returns the response bytes on success.
|
||||||
|
pub async fn send(
|
||||||
|
&self,
|
||||||
|
req: RequestBuilder,
|
||||||
|
) -> Result<Vec<u8>, RunError<HttpError>> {
|
||||||
|
let span = tracing::info_span!("resilient_http_send");
|
||||||
|
let op = move || {
|
||||||
|
let req = req.try_clone().unwrap();
|
||||||
|
let fut: Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, HttpError>> + Send>> =
|
||||||
|
Box::pin(async move {
|
||||||
|
let resp = req.send().instrument(tracing::trace_span!("http_send")).await?;
|
||||||
|
let bytes = resp.bytes().await?;
|
||||||
|
Ok::<Vec<u8>, HttpError>(bytes.to_vec())
|
||||||
|
});
|
||||||
|
fut
|
||||||
|
};
|
||||||
|
self.builder
|
||||||
|
.run(op)
|
||||||
|
.instrument(span)
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Convenience: GET `url`, returning response bytes.
|
||||||
|
pub async fn get(&self, url: &str) -> Result<Vec<u8>, RunError<HttpError>> {
|
||||||
|
self.send(self.inner.get(url)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Convenience: POST `url`, returning response bytes.
|
||||||
|
pub async fn post(&self, url: &str, body: Vec<u8>) -> Result<Vec<u8>, RunError<HttpError>> {
|
||||||
|
self.send(self.inner.post(url).body(body)).await
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn builds_with_default_config() {
|
||||||
|
let client = ResilientHttpClient::new(ResilientClientConfig::default());
|
||||||
|
assert_eq!(client.timeout(), Duration::from_secs(30));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn config_default_values() {
|
||||||
|
let c = ResilientClientConfig::default();
|
||||||
|
assert_eq!(c.timeout, Duration::from_secs(30));
|
||||||
|
assert_eq!(c.max_attempts, 1);
|
||||||
|
assert!(c.circuit_breaker.is_none());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-queue"
|
name = "mytheclipse-queue"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-storage"
|
name = "mytheclipse-storage"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-tracing"
|
name = "mytheclipse-tracing"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse"
|
name = "mytheclipse"
|
||||||
version = "1.10.0"
|
version = "1.12.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -34,6 +34,10 @@ pub mod bg;
|
|||||||
#[cfg(feature = "resiliency")]
|
#[cfg(feature = "resiliency")]
|
||||||
pub mod retry;
|
pub mod retry;
|
||||||
#[cfg(feature = "resiliency")]
|
#[cfg(feature = "resiliency")]
|
||||||
|
pub mod retry_ext;
|
||||||
|
#[cfg(feature = "resiliency")]
|
||||||
|
pub use retry_ext::RetryExt;
|
||||||
|
#[cfg(feature = "resiliency")]
|
||||||
pub mod circuit_breaker;
|
pub mod circuit_breaker;
|
||||||
#[cfg(feature = "resiliency")]
|
#[cfg(feature = "resiliency")]
|
||||||
pub mod timeout;
|
pub mod timeout;
|
||||||
@@ -114,7 +118,7 @@ pub use health::{HealthCheck, HealthRegistry, HealthStatus};
|
|||||||
pub use leader::{InProcLeaderElection, LeaderElection};
|
pub use leader::{InProcLeaderElection, LeaderElection};
|
||||||
|
|
||||||
#[cfg(feature = "resiliency")]
|
#[cfg(feature = "resiliency")]
|
||||||
pub use service_builder::ServiceBuilder;
|
pub use service_builder::{RunError, ServiceBuilder, ServiceConfig};
|
||||||
|
|
||||||
#[cfg(feature = "lifecycle")]
|
#[cfg(feature = "lifecycle")]
|
||||||
pub use dlock::{DistributedLock, LockError, LockGuard, InProcLock};
|
pub use dlock::{DistributedLock, LockError, LockGuard, InProcLock};
|
||||||
|
|||||||
@@ -0,0 +1,72 @@
|
|||||||
|
//! Ergonomic retry extension trait (feature `resiliency`).
|
||||||
|
//!
|
||||||
|
//! [`RetryExt`] extends any [`std::future::Future<Output = Result<T, E>>`]
|
||||||
|
//! with a `.retry()` method that delegates to [`crate::retry::retry`].
|
||||||
|
|
||||||
|
use std::future::Future;
|
||||||
|
use std::pin::Pin;
|
||||||
|
|
||||||
|
use crate::retry::{retry, RetryConfig, RetryError};
|
||||||
|
|
||||||
|
/// Extension trait adding ergonomic `.retry()` to any fallible future.
|
||||||
|
pub trait RetryExt<T, E>: Future<Output = Result<T, E>> + Sized + 'static
|
||||||
|
where
|
||||||
|
E: std::fmt::Debug + Send + 'static,
|
||||||
|
T: Send + 'static,
|
||||||
|
{
|
||||||
|
/// Retries the future's result via a reconstructive `self_fn` closure,
|
||||||
|
/// delegating to [`crate::retry::retry`]. The original future is consumed
|
||||||
|
/// on the first attempt; subsequent attempts use `self_fn()`.
|
||||||
|
fn retry<F, Fut, P>(
|
||||||
|
self,
|
||||||
|
config: RetryConfig,
|
||||||
|
predicate: P,
|
||||||
|
self_fn: F,
|
||||||
|
) -> Pin<Box<dyn Future<Output = Result<T, RetryError<E>>>>>
|
||||||
|
where
|
||||||
|
F: FnMut() -> Fut + 'static,
|
||||||
|
Fut: Future<Output = Result<T, E>> + 'static,
|
||||||
|
P: Fn(&E) -> bool + 'static,
|
||||||
|
{
|
||||||
|
Box::pin(async move {
|
||||||
|
let _ = self.await;
|
||||||
|
retry(config, predicate, self_fn).await
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<Fut, T, E> RetryExt<T, E> for Fut
|
||||||
|
where
|
||||||
|
Fut: Future<Output = Result<T, E>> + Send + 'static,
|
||||||
|
E: std::fmt::Debug + Send + 'static,
|
||||||
|
T: Send + 'static,
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::time::Duration;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::sync::atomic::{AtomicU32, Ordering};
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn retry_ext_retries_then_succeeds() {
|
||||||
|
let attempts = Arc::new(AtomicU32::new(0));
|
||||||
|
let a = Arc::clone(&attempts);
|
||||||
|
let cfg = RetryConfig { max_attempts: 3, base_delay: Duration::from_millis(1), ..RetryConfig::default() };
|
||||||
|
|
||||||
|
let op = move || {
|
||||||
|
let a = Arc::clone(&a);
|
||||||
|
async move {
|
||||||
|
let n = a.fetch_add(1, Ordering::SeqCst);
|
||||||
|
if n < 2 { Err::<(), String>("transient".into()) } else { Ok(()) }
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let fut = async { Err::<(), String>("first".into()) };
|
||||||
|
let result = fut.retry(cfg, |_: &String| true, op).await;
|
||||||
|
assert!(result.is_ok());
|
||||||
|
assert_eq!(attempts.load(Ordering::SeqCst), 3);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user