rocketmq-protocol 1.0.0

Runtime-neutral RocketMQ wire protocol contracts
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

// Copyright 2023 The RocketMQ Rust Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use serde::Deserialize;
use serde::Serialize;

use crate::protocol::subscription::retry_policy::RetryPolicy;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CustomizedRetryPolicy {
    next: Vec<i64>,
}

impl CustomizedRetryPolicy {
    pub fn next(&self) -> &[i64] {
        &self.next
    }
}

impl Default for CustomizedRetryPolicy {
    fn default() -> Self {
        let next = vec![
            1_000,     // 1s
            5_000,     // 5s
            10_000,    // 10s
            30_000,    // 30s
            60_000,    // 1m
            120_000,   // 2m
            180_000,   // 3m
            240_000,   // 4m
            300_000,   // 5m
            360_000,   // 6m
            420_000,   // 7m
            480_000,   // 8m
            540_000,   // 9m
            600_000,   // 10m
            1_200_000, // 20m
            1_800_000, // 30m
            3_600_000, // 1h
            7_200_000, // 2h
        ];
        CustomizedRetryPolicy { next }
    }
}

impl RetryPolicy for CustomizedRetryPolicy {
    fn next_delay_duration(&self, reconsume_times: i32) -> i64 {
        let reconsume_times = reconsume_times.max(0) as usize;
        let index = (reconsume_times + 2).min(self.next.len() - 1);
        self.next[index]
    }
}

#[cfg(test)]
mod customized_retry_policy_tests {
    use super::*;

    #[test]
    fn default_creates_expected_sequence() {
        let policy = CustomizedRetryPolicy::default();
        let expected_sequence = [
            1_000, 5_000, 10_000, 30_000, 60_000, 120_000, 180_000, 240_000, 300_000, 360_000, 420_000, 480_000,
            540_000, 600_000, 1_200_000, 1_800_000, 3_600_000, 7_200_000,
        ];
        assert_eq!(policy.next(), expected_sequence);
    }

    #[test]
    fn next_delay_duration_handles_offsets_and_boundaries() {
        let policy = CustomizedRetryPolicy::default();
        let test_cases = [
            (-1, 10_000),
            (0, 10_000),
            (1, 30_000),
            (14, 3_600_000),
            (15, 7_200_000),
            (16, 7_200_000),
        ];

        for (reconsume_times, expected_delay) in test_cases {
            assert_eq!(
                policy.next_delay_duration(reconsume_times),
                expected_delay,
                "unexpected delay for reconsume_times={reconsume_times}"
            );
        }
    }
}