rocketmq_protocol/protocol/static_topic/
logic_queue_mapping_item.rs

1// Copyright 2026 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 cheetah_string::CheetahString;
16use serde::Deserialize;
17use serde::Serialize;
18
19#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
20pub struct LogicQueueMappingItem {
21    pub gen: i32, // Immutable
22    #[serde(rename = "queueId")]
23    pub queue_id: i32, // Immutable
24    pub bname: Option<CheetahString>, // Immutable
25    #[serde(rename = "logicOffset")]
26    pub logic_offset: i64, // Start of the logic offset, Important, can be changed by command only once
27    #[serde(rename = "startOffset")]
28    pub start_offset: i64, // Start of the physical offset, Should always be 0, Immutable
29    #[serde(rename = "endOffset")]
30    pub end_offset: i64, // End of the physical offset, Excluded, Default to -1, Mutable
31    #[serde(rename = "timeOfStart")]
32    pub time_of_start: i64, // Mutable, Reserved
33    #[serde(rename = "timeOfEnd")]
34    pub time_of_end: i64, // Mutable, Reserved
35}
36
37impl Default for LogicQueueMappingItem {
38    fn default() -> Self {
39        Self {
40            gen: 0,
41            queue_id: 0,
42            bname: None,
43            logic_offset: 0,
44            start_offset: 0,
45            end_offset: -1,
46            time_of_start: -1,
47            time_of_end: -1,
48        }
49    }
50}
51
52impl LogicQueueMappingItem {
53    pub fn compute_static_queue_offset_strictly(&self, physical_queue_offset: i64) -> i64 {
54        if physical_queue_offset > self.start_offset {
55            return self.logic_offset;
56        }
57        self.logic_offset + (physical_queue_offset - self.start_offset)
58    }
59
60    pub fn compute_static_queue_offset_loosely(&self, physical_queue_offset: i64) -> i64 {
61        // Consider the newly mapped item
62        if self.logic_offset < 0 {
63            return -1;
64        }
65        if physical_queue_offset < self.start_offset {
66            return self.logic_offset;
67        }
68        if self.end_offset >= self.start_offset && self.end_offset < physical_queue_offset {
69            return self.logic_offset + (self.end_offset - self.start_offset);
70        }
71        self.logic_offset + (physical_queue_offset - self.start_offset)
72    }
73
74    pub fn compute_physical_queue_offset(&self, static_queue_offset: i64) -> i64 {
75        self.start_offset + (static_queue_offset - self.logic_offset)
76    }
77
78    pub fn compute_offset_delta(&self) -> i64 {
79        self.logic_offset - self.start_offset
80    }
81
82    pub fn check_if_end_offset_decided(&self) -> bool {
83        self.end_offset > self.start_offset
84    }
85
86    pub fn compute_max_static_queue_offset(&self) -> i64 {
87        if self.end_offset >= self.start_offset {
88            self.logic_offset + self.end_offset - self.start_offset
89        } else {
90            self.logic_offset
91        }
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98
99    #[test]
100    fn logic_queue_mapping_item_default() {
101        let item = LogicQueueMappingItem::default();
102        assert_eq!(item.gen, 0);
103        assert_eq!(item.queue_id, 0);
104        assert!(item.bname.is_none());
105        assert_eq!(item.logic_offset, 0);
106        assert_eq!(item.start_offset, 0);
107        assert_eq!(item.end_offset, -1);
108        assert_eq!(item.time_of_start, -1);
109        assert_eq!(item.time_of_end, -1);
110    }
111
112    #[test]
113    fn logic_queue_mapping_item_serde() {
114        let item = LogicQueueMappingItem {
115            gen: 1,
116            queue_id: 2,
117            bname: Some(CheetahString::from("broker-a")),
118            logic_offset: 100,
119            start_offset: 0,
120            end_offset: 200,
121            time_of_start: 123456,
122            time_of_end: 789012,
123        };
124        let json = serde_json::to_string(&item).unwrap();
125        let expected = r#"{"gen":1,"queueId":2,"bname":"broker-a","logicOffset":100,"startOffset":0,"endOffset":200,"timeOfStart":123456,"timeOfEnd":789012}"#;
126        assert_eq!(json, expected);
127        let deserialized: LogicQueueMappingItem = serde_json::from_str(&json).unwrap();
128        assert_eq!(item, deserialized);
129    }
130
131    #[test]
132    fn compute_static_queue_offset_strictly() {
133        let item = LogicQueueMappingItem {
134            start_offset: 100,
135            logic_offset: 500,
136            ..Default::default()
137        };
138        assert_eq!(item.compute_static_queue_offset_strictly(150), 500);
139        assert_eq!(item.compute_static_queue_offset_strictly(50), 450);
140    }
141
142    #[test]
143    fn compute_static_queue_offset_loosely() {
144        let mut item = LogicQueueMappingItem {
145            logic_offset: -1,
146            ..Default::default()
147        };
148        assert_eq!(item.compute_static_queue_offset_loosely(100), -1);
149
150        item.logic_offset = 500;
151        item.start_offset = 100;
152        assert_eq!(item.compute_static_queue_offset_loosely(50), 500);
153
154        item.end_offset = 200;
155        assert_eq!(item.compute_static_queue_offset_loosely(250), 600);
156
157        assert_eq!(item.compute_static_queue_offset_loosely(150), 550);
158    }
159
160    #[test]
161    fn compute_physical_queue_offset() {
162        let item = LogicQueueMappingItem {
163            start_offset: 100,
164            logic_offset: 500,
165            ..Default::default()
166        };
167        assert_eq!(item.compute_physical_queue_offset(550), 150);
168    }
169
170    #[test]
171    fn compute_offset_delta() {
172        let item = LogicQueueMappingItem {
173            start_offset: 100,
174            logic_offset: 500,
175            ..Default::default()
176        };
177        assert_eq!(item.compute_offset_delta(), 400);
178    }
179
180    #[test]
181    fn check_if_end_offset_decided() {
182        let mut item = LogicQueueMappingItem {
183            start_offset: 100,
184            end_offset: 50,
185            ..Default::default()
186        };
187        assert!(!item.check_if_end_offset_decided());
188        item.end_offset = 150;
189        assert!(item.check_if_end_offset_decided());
190    }
191
192    #[test]
193    fn compute_max_static_queue_offset() {
194        let mut item = LogicQueueMappingItem {
195            logic_offset: 500,
196            start_offset: 100,
197            end_offset: 50,
198            ..Default::default()
199        };
200        assert_eq!(item.compute_max_static_queue_offset(), 500);
201        item.end_offset = 200;
202        assert_eq!(item.compute_max_static_queue_offset(), 600);
203    }
204}
rocketmq_protocol/protocol/static_topic/logic_queue_mapping_item.rs

rocketmq_protocol/protocol/static_topic/
logic_queue_mapping_item.rs
