rocketmq_protocol/protocol/subscription/
customized_retry_policy.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use serde::Deserialize;
16use serde::Serialize;
17
18use crate::protocol::subscription::retry_policy::RetryPolicy;
19
20#[derive(Debug, Clone, Serialize, Deserialize)]
21pub struct CustomizedRetryPolicy {
22    next: Vec<i64>,
23}
24
25impl CustomizedRetryPolicy {
26    pub fn next(&self) -> &[i64] {
27        &self.next
28    }
29}
30
31impl Default for CustomizedRetryPolicy {
32    fn default() -> Self {
33        let next = vec![
34            1_000,     // 1s
35            5_000,     // 5s
36            10_000,    // 10s
37            30_000,    // 30s
38            60_000,    // 1m
39            120_000,   // 2m
40            180_000,   // 3m
41            240_000,   // 4m
42            300_000,   // 5m
43            360_000,   // 6m
44            420_000,   // 7m
45            480_000,   // 8m
46            540_000,   // 9m
47            600_000,   // 10m
48            1_200_000, // 20m
49            1_800_000, // 30m
50            3_600_000, // 1h
51            7_200_000, // 2h
52        ];
53        CustomizedRetryPolicy { next }
54    }
55}
56
57impl RetryPolicy for CustomizedRetryPolicy {
58    fn next_delay_duration(&self, reconsume_times: i32) -> i64 {
59        let reconsume_times = reconsume_times.max(0) as usize;
60        let index = (reconsume_times + 2).min(self.next.len() - 1);
61        self.next[index]
62    }
63}
64
65#[cfg(test)]
66mod customized_retry_policy_tests {
67    use super::*;
68
69    #[test]
70    fn default_creates_expected_sequence() {
71        let policy = CustomizedRetryPolicy::default();
72        let expected_sequence = [
73            1_000, 5_000, 10_000, 30_000, 60_000, 120_000, 180_000, 240_000, 300_000, 360_000, 420_000, 480_000,
74            540_000, 600_000, 1_200_000, 1_800_000, 3_600_000, 7_200_000,
75        ];
76        assert_eq!(policy.next(), expected_sequence);
77    }
78
79    #[test]
80    fn next_delay_duration_handles_offsets_and_boundaries() {
81        let policy = CustomizedRetryPolicy::default();
82        let test_cases = [
83            (-1, 10_000),
84            (0, 10_000),
85            (1, 30_000),
86            (14, 3_600_000),
87            (15, 7_200_000),
88            (16, 7_200_000),
89        ];
90
91        for (reconsume_times, expected_delay) in test_cases {
92            assert_eq!(
93                policy.next_delay_duration(reconsume_times),
94                expected_delay,
95                "unexpected delay for reconsume_times={reconsume_times}"
96            );
97        }
98    }
99}
rocketmq_protocol/protocol/subscription/customized_retry_policy.rs

rocketmq_protocol/protocol/subscription/
customized_retry_policy.rs
