gasket 0.9.0

Staged Event-Driven Architecture (SEDA) framework
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use serde::{de::Visitor, Deserialize, Deserializer, Serialize};
use std::{ops::Mul, time::Duration};
use tracing::debug;

#[derive(Clone, Debug, Default)]
pub struct Retry(Option<usize>);

impl Retry {
    pub fn fresh() -> Self {
        Default::default()
    }

    pub fn next(self) -> Self {
        Self(Some(self.0.unwrap_or(0) + 1))
    }

    pub fn maxed(&self, policy: &Policy) -> bool {
        match &self.0 {
            Some(num) => *num >= policy.max_retries,
            None => false,
        }
    }

    pub fn dismissed(&self, policy: &Policy) -> bool {
        self.maxed(policy) && policy.dismissible
    }

    pub async fn wait_backoff(&self, policy: &Policy, cancel: tokio_util::sync::CancellationToken) {
        let num = match &self.0 {
            Some(x) => x,
            None => return,
        };

        let backoff = compute_backoff_delay(policy, *num);

        debug!(
            "backoff for {}s until next retry #{}",
            backoff.as_secs(),
            num
        );

        tokio::select! {
            _ = cancel.cancelled() => (),
            _ = tokio::time::sleep(backoff) => ()
        }
    }
}

#[derive(Clone, Deserialize, Serialize, Default, Debug)]
pub struct Policy {
    pub max_retries: usize,
    #[serde(rename(deserialize = "backoff_unit_sec"))]
    #[serde(deserialize_with = "deserialize_duration")]
    pub backoff_unit: Duration,
    pub backoff_factor: u32,
    #[serde(rename(deserialize = "max_backoff_sec"))]
    #[serde(deserialize_with = "deserialize_duration")]
    pub max_backoff: Duration,
    pub dismissible: bool,
}

impl Policy {
    pub fn no_retry() -> Self {
        Self {
            max_retries: 0,
            backoff_unit: Default::default(),
            backoff_factor: Default::default(),
            max_backoff: Default::default(),
            dismissible: Default::default(),
        }
    }
}

fn compute_backoff_delay(policy: &Policy, retry: usize) -> Duration {
    let units = policy.backoff_factor.pow(retry as u32);
    let backoff = policy.backoff_unit.mul(units);
    core::cmp::min(backoff, policy.max_backoff)
}

fn deserialize_duration<'de, D>(deserializer: D) -> Result<Duration, D::Error>
where
    D: Deserializer<'de>,
{
    deserializer.deserialize_map(DurationVisitor)
}

struct DurationVisitor;
impl<'de> Visitor<'de> for DurationVisitor {
    type Value = Duration;

    fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
        formatter.write_str("This Visitor expects to receive i64 seconds")
    }

    fn visit_i64<E>(self, v: i64) -> Result<Self::Value, E>
    where
        E: serde::de::Error,
    {
        Ok(Duration::from_secs(v as u64))
    }
}