rocketmq_protocol/protocol/
data_version_compat.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//! Clock-owning compatibility helpers for the canonical data-version value.
10
11use std::time::SystemTime;
12use std::time::UNIX_EPOCH;
13
14use crate::protocol::DataVersion;
15
16pub fn new_data_version() -> DataVersion {
17    DataVersion::with_values(0, current_millis(), 0)
18}
19
20pub trait DataVersionExt {
21    fn next_version(&mut self);
22    fn next_version_with(&mut self, state_version: i64);
23}
24
25impl DataVersionExt for DataVersion {
26    fn next_version(&mut self) {
27        self.next_version_with(0);
28    }
29
30    fn next_version_with(&mut self, state_version: i64) {
31        self.next_version_with_timestamp(state_version, current_millis());
32    }
33}
34
35fn current_millis() -> i64 {
36    SystemTime::now()
37        .duration_since(UNIX_EPOCH)
38        .map(|duration| duration.as_millis() as i64)
39        .unwrap_or(0)
40}
41
42#[cfg(test)]
43mod tests {
44    use super::current_millis;
45    use super::new_data_version;
46    use super::DataVersionExt;
47
48    #[test]
49    fn new_data_version_starts_at_zero_with_wall_clock_timestamp() {
50        let before = current_millis();
51        let version = new_data_version();
52        let after = current_millis();
53
54        assert_eq!(version.state_version(), 0);
55        assert_eq!(version.counter(), 0);
56        assert!(version.timestamp() > 0);
57        assert!((before..=after).contains(&version.timestamp()));
58    }
59
60    #[test]
61    fn next_version_bumps_counter_and_keeps_state_version_zero() {
62        let mut version = new_data_version();
63        let previous_timestamp = version.timestamp();
64
65        version.next_version();
66
67        assert_eq!(version.counter(), 1);
68        assert_eq!(version.state_version(), 0);
69        assert!(version.timestamp() >= previous_timestamp);
70        assert!(version.timestamp() <= current_millis());
71    }
72
73    #[test]
74    fn next_version_with_sets_state_version_and_bumps_counter() {
75        let mut version = new_data_version();
76        let previous_timestamp = version.timestamp();
77
78        version.next_version_with(7);
79
80        assert_eq!(version.state_version(), 7);
81        assert_eq!(version.counter(), 1);
82        assert!(version.timestamp() >= previous_timestamp);
83    }
84
85    #[test]
86    fn repeated_version_helpers_yield_strictly_increasing_counters() {
87        let mut version = new_data_version();
88        let mut previous_counter = version.counter();
89
90        for state_version in 1..=5 {
91            version.next_version();
92            assert!(version.counter() > previous_counter);
93            previous_counter = version.counter();
94
95            version.next_version_with(state_version);
96            assert!(version.counter() > previous_counter);
97            assert_eq!(version.state_version(), state_version);
98            previous_counter = version.counter();
99        }
100
101        assert_eq!(version.counter(), 10);
102    }
103}
rocketmq_protocol/protocol/data_version_compat.rs

rocketmq_protocol/protocol/
data_version_compat.rs
