rocketmq_protocol/protocol/static_topic/
topic_queue_mapping_context.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 cheetah_string::CheetahString;
16
17use crate::protocol::static_topic::logic_queue_mapping_item::LogicQueueMappingItem;
18use crate::protocol::static_topic::topic_queue_mapping_detail::TopicQueueMappingDetail;
19
20#[derive(Default)]
21pub struct TopicQueueMappingContext {
22    pub topic: CheetahString,
23    pub global_id: Option<i32>,
24    pub mapping_detail: Option<TopicQueueMappingDetail>,
25    pub mapping_item_list: Vec<LogicQueueMappingItem>,
26    pub leader_item: Option<LogicQueueMappingItem>,
27    pub current_item: Option<LogicQueueMappingItem>,
28}
29
30impl TopicQueueMappingContext {
31    pub fn new(
32        topic: impl Into<CheetahString>,
33        global_id: Option<i32>,
34        mapping_detail: Option<TopicQueueMappingDetail>,
35        mapping_item_list: Vec<LogicQueueMappingItem>,
36        leader_item: Option<LogicQueueMappingItem>,
37    ) -> Self {
38        Self {
39            topic: topic.into(),
40            global_id,
41            mapping_detail,
42            mapping_item_list,
43            leader_item,
44            current_item: None,
45        }
46    }
47
48    pub fn is_leader(&self) -> bool {
49        if let Some(leader_item) = &self.leader_item {
50            match leader_item.bname {
51                None => {
52                    return false;
53                }
54                Some(ref broker_name) => match self.mapping_detail.as_ref() {
55                    None => return false,
56                    Some(inner) => match inner.topic_queue_mapping_info.bname {
57                        None => return false,
58                        Some(ref inner_bname) => {
59                            return inner_bname == broker_name;
60                        }
61                    },
62                },
63            }
64        }
65        false
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72    use crate::protocol::static_topic::topic_queue_mapping_info::TopicQueueMappingInfo;
73
74    #[test]
75    fn topic_queue_mapping_context_default() {
76        let context = TopicQueueMappingContext::default();
77        assert_eq!(context.topic, "");
78        assert!(context.global_id.is_none());
79        assert!(context.mapping_detail.is_none());
80        assert!(context.mapping_item_list.is_empty());
81        assert!(context.leader_item.is_none());
82        assert!(context.current_item.is_none());
83    }
84
85    #[test]
86    fn topic_queue_mapping_context_new() {
87        let topic = "test_topic";
88        let global_id = Some(1);
89        let mapping_detail = TopicQueueMappingDetail::default();
90        let mapping_item_list = vec![LogicQueueMappingItem::default()];
91        let leader_item = LogicQueueMappingItem::default();
92
93        let context = TopicQueueMappingContext::new(
94            topic,
95            global_id,
96            Some(mapping_detail.clone()),
97            mapping_item_list.clone(),
98            Some(leader_item.clone()),
99        );
100
101        assert_eq!(context.topic, CheetahString::from(topic));
102        assert_eq!(context.global_id, global_id);
103        assert_eq!(context.mapping_detail.unwrap(), mapping_detail);
104        assert_eq!(context.mapping_item_list.len(), 1);
105        assert_eq!(context.leader_item.unwrap(), leader_item);
106        assert!(context.current_item.is_none());
107    }
108
109    #[test]
110    fn topic_queue_mapping_context_is_leader() {
111        let broker_name = CheetahString::from("broker_a");
112        let leader_item = LogicQueueMappingItem {
113            bname: Some(broker_name.clone()),
114            ..Default::default()
115        };
116        let mapping_detail = TopicQueueMappingDetail {
117            topic_queue_mapping_info: TopicQueueMappingInfo {
118                bname: Some(broker_name.clone()),
119                ..TopicQueueMappingInfo::default()
120            },
121            ..Default::default()
122        };
123        let mut context = TopicQueueMappingContext {
124            leader_item: Some(leader_item.clone()),
125            mapping_detail: Some(mapping_detail),
126            ..Default::default()
127        };
128
129        // Match: should be leader
130        assert!(context.is_leader());
131
132        // No mapping_detail: not leader
133        context.mapping_detail = None;
134        assert!(!context.is_leader());
135
136        // Different bname in mapping_detail: not leader
137        let diff_mapping_detail = TopicQueueMappingDetail {
138            topic_queue_mapping_info: TopicQueueMappingInfo {
139                bname: Some(CheetahString::from("broker_b")),
140                ..TopicQueueMappingInfo::default()
141            },
142            ..Default::default()
143        };
144        context.mapping_detail = Some(diff_mapping_detail);
145        assert!(!context.is_leader());
146
147        // No bname in leader_item: not leader
148        context.leader_item.as_mut().unwrap().bname = None;
149        assert!(!context.is_leader());
150
151        // No leader_item: not leader
152        context.leader_item = None;
153        assert!(!context.is_leader());
154    }
155}
rocketmq_protocol/protocol/static_topic/topic_queue_mapping_context.rs

rocketmq_protocol/protocol/static_topic/
topic_queue_mapping_context.rs
