azeventhubs 0.20.0

An unofficial AMQP 1.0 rust client for Azure Event Hubs
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

use std::time::Duration;

use rand::Rng;

use crate::{
    event_hubs_retry_mode::EventHubsRetryMode, event_hubs_retry_options::EventHubsRetryOptions,
    event_hubs_retry_policy::EventHubsRetryPolicy,
};

const DEFAULT_JITTER_FACTOR: f64 = 0.08;
// const DEFAULT_MINIMUM_THROTTLE_SECONDS: u32 = 4;
// const DEFAULT_MAXIMUM_THROTTLE_SECONDS: u32 = 8;

/// Default retry policy used by the client.
#[derive(Debug, Clone)]
pub struct BasicRetryPolicy {
    options: EventHubsRetryOptions,
    jitter_factor: f64,
    // /// The minimum number of seconds to wait before retrying when the service
    // /// indicates that the client should throttle its requests.
    // ///
    // /// TODO: This is not being used right now
    // minimum_throttle_seconds: u32,

    // /// The maximum number of seconds to wait before retrying when the service
    // /// indicates that the client should throttle its requests.
    // ///
    // /// TODO: This is not being used right now
    // maximum_throttle_seconds: u32,
}

impl BasicRetryPolicy {
    /// The set of options used to configure the retry policy.
    pub fn options(&self) -> &EventHubsRetryOptions {
        &self.options
    }

    /// The factor to apply to the base delay for use as a base jitter value.
    pub fn jitter_factor(&self) -> f64 {
        self.jitter_factor
    }
}

impl From<EventHubsRetryOptions> for BasicRetryPolicy {
    fn from(options: EventHubsRetryOptions) -> Self {
        Self {
            options,
            jitter_factor: DEFAULT_JITTER_FACTOR,
            // minimum_throttle_seconds: DEFAULT_MINIMUM_THROTTLE_SECONDS,
            // maximum_throttle_seconds: DEFAULT_MAXIMUM_THROTTLE_SECONDS,
        }
    }
}

impl EventHubsRetryPolicy for BasicRetryPolicy {
    fn calculate_try_timeout(&self, _attempt_count: u32) -> std::time::Duration {
        self.options.try_timeout
    }

    fn calculate_retry_delay(
        &self,
        _last_error: &dyn std::error::Error,
        attempt_count: u32,
    ) -> Option<std::time::Duration> {
        if self.options.delay == std::time::Duration::ZERO
            || self.options.maximum_delay == std::time::Duration::ZERO
            || attempt_count > self.options.max_retries.0
        // || !self.should_retry_last_error(last_error) // TODO:
        {
            return None;
        }

        let base_jitter_seconds = self.options.delay.as_secs_f64() * self.jitter_factor;
        let retry_delay = match &self.options.mode {
            EventHubsRetryMode::Fixed => {
                calculate_fixed_delay(self.options.delay.as_secs_f64(), base_jitter_seconds)
            }
            EventHubsRetryMode::Exponential => calculate_exponential_delay(
                self.options.delay.as_secs_f64(),
                base_jitter_seconds,
                attempt_count,
            ),
        };

        // TODO: determine if the error represents a request to throttle

        // Adjust the delay
        if retry_delay > self.options.maximum_delay {
            Some(self.options.maximum_delay)
        } else {
            Some(retry_delay)
        }
    }
}

fn calculate_fixed_delay(base_delay_seconds: f64, base_jitter_seconds: f64) -> Duration {
    let mut rng = rand::thread_rng();
    let delay = base_delay_seconds * (rng.gen_range(0.0..1.0) * base_jitter_seconds);
    if delay > Duration::MAX.as_secs_f64() {
        Duration::MAX
    } else {
        Duration::from_secs_f64(delay)
    }
}

fn calculate_exponential_delay(
    base_delay_seconds: f64,
    base_jitter_seconds: f64,
    attempt_count: u32,
) -> Duration {
    let mut rng = rand::thread_rng();
    let delay = (f64::powi(2.0, attempt_count as i32) * base_delay_seconds)
        + (rng.gen_range(0.0..1.0) * base_jitter_seconds);
    if delay > Duration::MAX.as_secs_f64() {
        Duration::MAX
    } else {
        Duration::from_secs_f64(delay)
    }
}