crius 0.4.0

Crius is a simple hystrix-like circuit breaker
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

use error::CriusError;
use circuit_breaker::CircuitBreaker;
use std::marker::PhantomData;

const DEFAULT_ERROR_THRESHOLD: i32 = 10;
const DEFAULT_ERROR_THRESHOLD_PERCENTAGE: i32 = 50;
const DEFAULT_BUCKETS_IN_WINDOW: u32 = 10;
const DEFAULT_BUCKET_SIZE_IN_MS: u64 = 1000;
const DEFAULT_CIRCUIT_OPEN_MS: u64 = 5000;
const DEFAULT_CIRCUIT_BREAKER_ENABLED: bool = true;

#[derive(Copy, Clone, Debug)]
pub struct Config {
    pub error_threshold: i32,
    pub error_threshold_percentage: i32,
    pub buckets_in_window: u32,
    pub bucket_size_in_ms: u64,
    pub circuit_open_ms: u64,
    pub circuit_breaker_enabled: bool,
}

impl Config {
    pub fn default() -> Config {
        Config {
            error_threshold: DEFAULT_ERROR_THRESHOLD,
            error_threshold_percentage: DEFAULT_ERROR_THRESHOLD_PERCENTAGE,
            buckets_in_window: DEFAULT_BUCKETS_IN_WINDOW,
            bucket_size_in_ms: DEFAULT_BUCKET_SIZE_IN_MS,
            circuit_open_ms: DEFAULT_CIRCUIT_OPEN_MS,
            circuit_breaker_enabled: DEFAULT_CIRCUIT_BREAKER_ENABLED,
        }
    }

    pub fn error_threshold(&mut self, error_threshold: i32) -> &mut Self {
        self.error_threshold = error_threshold;
        return self;
    }

    pub fn error_threshold_percentage(&mut self, error_threshold_percentage: i32) -> &mut Self {
        self.error_threshold_percentage = error_threshold_percentage;
        return self;
    }

    pub fn buckets_in_window(&mut self, buckets_in_window: u32) -> &mut Self {
        self.buckets_in_window = buckets_in_window;
        return self;
    }

    pub fn bucket_size_in_ms(&mut self, bucket_size_in_ms: u64) -> &mut Self {
        self.bucket_size_in_ms = bucket_size_in_ms;
        return self;
    }

    pub fn circuit_open_ms(&mut self, circuit_open_ms: u64) -> &mut Self {
        self.circuit_open_ms = circuit_open_ms;
        return self;
    }

    pub fn circuit_breaker_enabled(&mut self, circuit_breaker_enabled: bool) -> &mut Self {
        self.circuit_breaker_enabled = circuit_breaker_enabled;
        return self;
    }
}

pub struct Command<I, O, E, F, FB>
where
    E: From<CriusError>,
    F: Fn(I) -> Result<O, E>,
    FB: Fn(E) -> O,
{
    pub cmd: F,
    pub fallback: Option<FB>,
    phantom_data: PhantomData<I>,
    circuit_breaker: CircuitBreaker,
}

impl<I, O, E, F, FB> Command<I, O, E, F, FB>
where
    E: From<CriusError>,
    F: Fn(I) -> Result<O, E>,
    FB: Fn(E) -> O,
{
    pub fn define(cfg: Config, cmd: F) -> Result<Command<I, O, E, F, FB>, CriusError> {
        Ok(Command {
            cmd: cmd,
            fallback: None,
            phantom_data: PhantomData,
            circuit_breaker: CircuitBreaker::new(cfg)?,
        })
    }

    pub fn define_with_fallback(
        cfg: Config,
        cmd: F,
        fallback: FB,
    ) -> Result<Command<I, O, E, F, FB>, CriusError> {
        Ok(Command {
            cmd: cmd,
            fallback: Some(fallback),
            phantom_data: PhantomData,
            circuit_breaker: CircuitBreaker::new(cfg)?,
        })
    }

    pub fn run(&mut self, param: I) -> Result<O, E> {
        // Run the command if the breaker is disabled:
        let enabled = self.circuit_breaker.config.circuit_breaker_enabled;
        if !enabled {
            return (self.cmd)(param);
        }

        // Execute the command if the breaker is enabled and execution
        // is allowed.
        let is_allowed = self.circuit_breaker.check_command_allowed();
        if is_allowed {
            let result = (self.cmd)(param);
            self.circuit_breaker.register_result(&result);

            return match result {
                Ok(result) => Ok(result),
                Err(err) => {
                    // If a fallback is configured, use it on error:
                    if let Some(ref fallback) = self.fallback {
                        Ok(fallback(err))
                    } else {
                        Err(err)
                    }
                }
            };
        }

        // If execution is rejected, either run the configured
        // fallback (if present) or propagate the rejection as an
        // error:
        let err = E::from(CriusError::ExecutionRejected);
        if let Some(ref fallback) = self.fallback {
            return Ok(fallback(err));
        } else {
            return Err(err);
        }
    }
}