rocketmq_protocol/protocol/body/
epoch_entry_cache.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 std::fmt;
16
17use cheetah_string::CheetahString;
18use serde::Deserialize;
19use serde::Serialize;
20
21#[derive(Deserialize, Serialize, Debug, Clone, PartialEq, Eq)]
22#[serde(rename_all = "camelCase")]
23pub struct EpochEntry {
24    #[serde(default)]
25    epoch: i32,
26    #[serde(default)]
27    start_offset: i64,
28    #[serde(default = "EpochEntry::default_end_offset")]
29    end_offset: i64,
30}
31
32impl Default for EpochEntry {
33    fn default() -> Self {
34        Self::new(0, 0)
35    }
36}
37
38impl EpochEntry {
39    fn default_end_offset() -> i64 {
40        i64::MAX
41    }
42
43    pub fn new(epoch: i32, start_offset: i64) -> Self {
44        Self {
45            epoch,
46            start_offset,
47            end_offset: i64::MAX,
48        }
49    }
50
51    pub fn with_end_offset(epoch: i32, start_offset: i64, end_offset: i64) -> Self {
52        Self {
53            epoch,
54            start_offset,
55            end_offset,
56        }
57    }
58
59    pub fn get_epoch(&self) -> i32 {
60        self.epoch
61    }
62
63    pub fn set_epoch(&mut self, epoch: i32) {
64        self.epoch = epoch;
65    }
66
67    pub fn get_start_offset(&self) -> i64 {
68        self.start_offset
69    }
70
71    pub fn set_start_offset(&mut self, start_offset: i64) {
72        self.start_offset = start_offset;
73    }
74
75    pub fn get_end_offset(&self) -> i64 {
76        self.end_offset
77    }
78
79    pub fn set_end_offset(&mut self, end_offset: i64) {
80        self.end_offset = end_offset;
81    }
82}
83
84impl fmt::Display for EpochEntry {
85    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86        write!(
87            f,
88            "EpochEntry{{epoch={}, startOffset={}, endOffset={}}}",
89            self.epoch, self.start_offset, self.end_offset
90        )
91    }
92}
93
94#[derive(Deserialize, Serialize, Default)]
95#[serde(rename_all = "camelCase")]
96pub struct EpochEntryCache {
97    cluster_name: CheetahString,
98    broker_name: CheetahString,
99    broker_id: u64,
100    epoch_list: Vec<EpochEntry>,
101    max_offset: u64,
102}
103
104impl EpochEntryCache {
105    pub fn new(
106        cluster_name: impl Into<CheetahString>,
107        broker_name: impl Into<CheetahString>,
108        broker_id: u64,
109        epoch_list: Vec<EpochEntry>,
110        max_offset: u64,
111    ) -> Self {
112        Self {
113            cluster_name: cluster_name.into(),
114            broker_name: broker_name.into(),
115            broker_id,
116            epoch_list,
117            max_offset,
118        }
119    }
120    pub fn get_cluster_name(&self) -> &CheetahString {
121        &self.cluster_name
122    }
123    pub fn set_cluster_name(&mut self, cluster_name: impl Into<CheetahString>) {
124        self.cluster_name = cluster_name.into()
125    }
126    pub fn get_broker_name(&self) -> &CheetahString {
127        &self.broker_name
128    }
129    pub fn set_broker_name(&mut self, broker_name: impl Into<CheetahString>) {
130        self.broker_name = broker_name.into()
131    }
132    pub fn get_broker_id(&self) -> u64 {
133        self.broker_id
134    }
135    pub fn get_epoch_list(&self) -> &Vec<EpochEntry> {
136        &self.epoch_list
137    }
138    pub fn get_epoch_list_mut(&mut self) -> &mut Vec<EpochEntry> {
139        &mut self.epoch_list
140    }
141    pub fn get_max_offset(&self) -> u64 {
142        self.max_offset
143    }
144    pub fn set_max_offset(&mut self, max_offset: u64) {
145        self.max_offset = max_offset;
146    }
147}
148
149#[cfg(test)]
150mod tests {
151    use cheetah_string::CheetahString;
152
153    use super::*;
154
155    #[test]
156    fn epoch_entry_default_matches_constructor_and_missing_wire_fields() {
157        let expected = EpochEntry::new(0, 0);
158        assert_eq!(expected.get_end_offset(), i64::MAX);
159        assert_eq!(EpochEntry::default(), expected);
160        assert_eq!(serde_json::from_str::<EpochEntry>("{}").unwrap(), expected);
161        assert_eq!(
162            serde_json::to_value(&expected).unwrap(),
163            serde_json::json!({
164                "epoch": 0, "startOffset": 0, "endOffset": i64::MAX
165            })
166        );
167        for end_offset in [0, 42] {
168            let value = serde_json::json!({"endOffset": end_offset});
169            assert_eq!(
170                serde_json::from_value::<EpochEntry>(value).unwrap(),
171                EpochEntry::with_end_offset(0, 0, end_offset)
172            );
173        }
174    }
175
176    #[test]
177    fn epoch_entry_and_cache_accessors_update_all_fields() {
178        let mut entry = EpochEntry::new(1, 0);
179        assert_eq!(entry.get_end_offset(), i64::MAX);
180        entry.set_epoch(2);
181        entry.set_start_offset(100);
182        entry.set_end_offset(200);
183        assert_eq!(entry.get_epoch(), 2);
184        assert_eq!(entry.get_start_offset(), 100);
185        assert_eq!(entry.get_end_offset(), 200);
186
187        let mut cache = EpochEntryCache::new("cluster1", "broker1", 1, vec![entry], 500);
188        assert_eq!(cache.get_cluster_name(), &CheetahString::from("cluster1"));
189        assert_eq!(cache.get_broker_name(), &CheetahString::from("broker1"));
190        assert_eq!(cache.get_broker_id(), 1);
191        assert_eq!(cache.get_epoch_list().len(), 1);
192        assert_eq!(cache.get_max_offset(), 500);
193
194        cache.set_cluster_name("cluster2");
195        cache.set_broker_name("broker2");
196        cache.set_max_offset(600);
197        cache.get_epoch_list_mut().push(EpochEntry::new(3, 200));
198        assert_eq!(cache.get_cluster_name(), &CheetahString::from("cluster2"));
199        assert_eq!(cache.get_broker_name(), &CheetahString::from("broker2"));
200        assert_eq!(cache.get_max_offset(), 600);
201        assert_eq!(cache.get_epoch_list().len(), 2);
202    }
203
204    #[test]
205    fn epoch_entry_display() {
206        let entry = EpochEntry::with_end_offset(1, 100, 200);
207        assert_eq!(entry.to_string(), "EpochEntry{epoch=1, startOffset=100, endOffset=200}");
208    }
209}
rocketmq_protocol/protocol/body/epoch_entry_cache.rs

rocketmq_protocol/protocol/body/
epoch_entry_cache.rs
