rocketmq_common/common/
config.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::collections::HashMap;
16use std::fmt::Display;
17
18use cheetah_string::CheetahString;
19use serde::Deserialize;
20use serde::Serialize;
21
22use super::TopicFilterType;
23use crate::common::attribute::topic_message_type::TopicMessageType;
24use crate::common::attribute::Attribute;
25use crate::common::constant::PermName;
26use crate::TopicAttributes::TopicAttributes;
27
28#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
29#[serde(rename_all = "camelCase")]
30pub struct TopicConfig {
31    pub topic_name: Option<CheetahString>,
32    pub read_queue_nums: u32,
33    pub write_queue_nums: u32,
34    pub perm: u32,
35    pub topic_filter_type: TopicFilterType,
36    pub topic_sys_flag: u32,
37    pub order: bool,
38    pub attributes: HashMap<CheetahString, CheetahString>,
39}
40
41impl Display for TopicConfig {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        write!(
44            f,
45            "TopicConfig {{ topic_name: {:?}, read_queue_nums: {}, write_queue_nums: {}, perm: {}, topic_filter_type: \
46             {}, topic_sys_flag: {}, order: {}, attributes: {:?} }}",
47            self.topic_name,
48            self.read_queue_nums,
49            self.write_queue_nums,
50            self.perm,
51            self.topic_filter_type,
52            self.topic_sys_flag,
53            self.order,
54            self.attributes
55        )
56    }
57}
58
59impl Default for TopicConfig {
60    fn default() -> Self {
61        Self {
62            topic_name: None,
63            read_queue_nums: Self::DEFAULT_READ_QUEUE_NUMS,
64            write_queue_nums: Self::DEFAULT_WRITE_QUEUE_NUMS,
65            perm: PermName::PERM_READ | PermName::PERM_WRITE,
66            topic_filter_type: TopicFilterType::SingleTag,
67            topic_sys_flag: 0,
68            order: false,
69            attributes: HashMap::new(),
70        }
71    }
72}
73
74impl TopicConfig {
75    const DEFAULT_READ_QUEUE_NUMS: u32 = 16;
76    const DEFAULT_WRITE_QUEUE_NUMS: u32 = 16;
77    const SEPARATOR: &'static str = " ";
78
79    pub fn get_topic_message_type(&self) -> TopicMessageType {
80        if self.attributes.is_empty() {
81            return TopicMessageType::Normal;
82        }
83        let content = self
84            .attributes
85            .get(TopicAttributes::topic_message_type_attribute().name());
86        if let Some(content) = content {
87            return TopicMessageType::from(content.to_string());
88        }
89        TopicMessageType::Normal
90    }
91
92    pub fn set_lite_topic_expiration(&mut self, lite_topic_expiration: i32) {
93        if self.get_topic_message_type() != TopicMessageType::Lite {
94            return;
95        }
96        self.attributes.insert(
97            TopicAttributes::lite_topic_expiration_attribute().name().clone(),
98            CheetahString::from_string(lite_topic_expiration.to_string()),
99        );
100    }
101
102    pub fn get_lite_topic_expiration(&self) -> i32 {
103        if self.get_topic_message_type() != TopicMessageType::Lite {
104            return -1;
105        }
106        self.attributes
107            .get(TopicAttributes::lite_topic_expiration_attribute().name())
108            .and_then(|content| content.parse::<i32>().ok())
109            .unwrap_or(-1)
110    }
111
112    pub fn new(topic_name: impl Into<CheetahString>) -> Self {
113        TopicConfig {
114            topic_name: Some(topic_name.into()),
115            ..Self::default()
116        }
117    }
118
119    pub fn with_queues(topic_name: impl Into<CheetahString>, read_queue_nums: u32, write_queue_nums: u32) -> Self {
120        Self {
121            read_queue_nums,
122            write_queue_nums,
123            ..Self::new(topic_name)
124        }
125    }
126
127    pub fn with_perm(
128        topic_name: impl Into<CheetahString>,
129        read_queue_nums: u32,
130        write_queue_nums: u32,
131        perm: u32,
132    ) -> Self {
133        Self {
134            read_queue_nums,
135            write_queue_nums,
136            perm,
137            ..Self::new(topic_name)
138        }
139    }
140
141    pub fn with_sys_flag(
142        topic_name: impl Into<CheetahString>,
143        read_queue_nums: u32,
144        write_queue_nums: u32,
145        perm: u32,
146        topic_sys_flag: u32,
147    ) -> Self {
148        Self {
149            read_queue_nums,
150            write_queue_nums,
151            perm,
152            topic_sys_flag,
153            ..Self::new(topic_name)
154        }
155    }
156
157    pub fn encode(&self) -> String {
158        let mut sb = String::new();
159        sb.push_str(self.topic_name.clone().unwrap_or(CheetahString::empty()).as_str());
160        sb.push_str(Self::SEPARATOR);
161        sb.push_str(&self.read_queue_nums.to_string());
162        sb.push_str(Self::SEPARATOR);
163        sb.push_str(&self.write_queue_nums.to_string());
164        sb.push_str(Self::SEPARATOR);
165        sb.push_str(&self.perm.to_string());
166        sb.push_str(Self::SEPARATOR);
167        sb.push_str(&format!("{}", self.topic_filter_type));
168        sb.push_str(Self::SEPARATOR);
169        if !self.attributes.is_empty() {
170            append_json_string_map(&mut sb, &self.attributes);
171        }
172        sb.trim().to_string()
173    }
174
175    pub fn decode(&mut self, input: &str) -> bool {
176        let parts: Vec<&str> = input.split(Self::SEPARATOR).collect();
177        if parts.len() >= 5 {
178            self.topic_name = Some(parts[0].into());
179            self.read_queue_nums = parts[1].parse().unwrap_or(Self::DEFAULT_READ_QUEUE_NUMS);
180            self.write_queue_nums = parts[2].parse().unwrap_or(Self::DEFAULT_WRITE_QUEUE_NUMS);
181            self.perm = parts[3].parse().unwrap_or(PermName::PERM_READ | PermName::PERM_WRITE);
182            self.topic_filter_type = From::from(parts[4]);
183            if parts.len() >= 6 {
184                if let Ok(attrs) = serde_json::from_str(parts[5]) {
185                    self.attributes = attrs
186                }
187            }
188            true
189        } else {
190            false
191        }
192    }
193
194    pub fn get_read_queue_nums(&self) -> u32 {
195        self.read_queue_nums
196    }
197}
198
199fn append_json_string_map(output: &mut String, values: &HashMap<CheetahString, CheetahString>) {
200    output.push('{');
201    for (index, (key, value)) in values.iter().enumerate() {
202        if index > 0 {
203            output.push(',');
204        }
205        append_json_string(output, key.as_str());
206        output.push(':');
207        append_json_string(output, value.as_str());
208    }
209    output.push('}');
210}
211
212fn append_json_string(output: &mut String, value: &str) {
213    output.push('"');
214    for ch in value.chars() {
215        match ch {
216            '"' => output.push_str("\\\""),
217            '\\' => output.push_str("\\\\"),
218            '\u{08}' => output.push_str("\\b"),
219            '\u{0C}' => output.push_str("\\f"),
220            '\n' => output.push_str("\\n"),
221            '\r' => output.push_str("\\r"),
222            '\t' => output.push_str("\\t"),
223            ch if ch <= '\u{1F}' => {
224                output.push_str("\\u00");
225                const HEX: &[u8; 16] = b"0123456789abcdef";
226                let byte = ch as u8;
227                output.push(HEX[(byte >> 4) as usize] as char);
228                output.push(HEX[(byte & 0x0f) as usize] as char);
229            }
230            ch => output.push(ch),
231        }
232    }
233    output.push('"');
234}
235
236#[cfg(test)]
237mod tests {
238    use std::collections::HashMap;
239
240    use cheetah_string::CheetahString;
241
242    use super::*;
243
244    #[test]
245    fn default_topic_config() {
246        let config = TopicConfig::default();
247        assert_eq!(config.topic_name, None);
248        assert_eq!(config.read_queue_nums, TopicConfig::DEFAULT_READ_QUEUE_NUMS);
249        assert_eq!(config.write_queue_nums, TopicConfig::DEFAULT_WRITE_QUEUE_NUMS);
250        assert_eq!(config.perm, PermName::PERM_READ | PermName::PERM_WRITE);
251        assert_eq!(config.topic_filter_type, TopicFilterType::SingleTag);
252        assert_eq!(config.topic_sys_flag, 0);
253        assert!(!config.order);
254        assert!(config.attributes.is_empty());
255    }
256
257    #[test]
258    fn new_topic_config() {
259        let topic_name = CheetahString::from("test_topic");
260        let config = TopicConfig::new(topic_name.clone());
261        assert_eq!(config.topic_name, Some(topic_name));
262    }
263
264    #[test]
265    fn with_queues_topic_config() {
266        let topic_name = CheetahString::from("test_topic");
267        let config = TopicConfig::with_queues(topic_name.clone(), 8, 8);
268        assert_eq!(config.topic_name, Some(topic_name));
269        assert_eq!(config.read_queue_nums, 8);
270        assert_eq!(config.write_queue_nums, 8);
271    }
272
273    #[test]
274    fn with_perm_topic_config() {
275        let topic_name = CheetahString::from("test_topic");
276        let config = TopicConfig::with_perm(topic_name.clone(), 8, 8, PermName::PERM_READ);
277        assert_eq!(config.topic_name, Some(topic_name));
278        assert_eq!(config.read_queue_nums, 8);
279        assert_eq!(config.write_queue_nums, 8);
280        assert_eq!(config.perm, PermName::PERM_READ);
281    }
282
283    #[test]
284    fn with_sys_flag_topic_config() {
285        let topic_name = CheetahString::from("test_topic");
286        let config = TopicConfig::with_sys_flag(topic_name.clone(), 8, 8, PermName::PERM_READ, 1);
287        assert_eq!(config.topic_name, Some(topic_name));
288        assert_eq!(config.read_queue_nums, 8);
289        assert_eq!(config.write_queue_nums, 8);
290        assert_eq!(config.perm, PermName::PERM_READ);
291        assert_eq!(config.topic_sys_flag, 1);
292    }
293
294    #[test]
295    fn encode_topic_config() {
296        let topic_name = CheetahString::from("test_topic");
297        let mut attributes = HashMap::new();
298        attributes.insert(CheetahString::from("key"), CheetahString::from("value"));
299        let config = TopicConfig {
300            topic_name: Some(topic_name.clone()),
301            read_queue_nums: 8,
302            write_queue_nums: 8,
303            perm: PermName::PERM_READ,
304            topic_filter_type: TopicFilterType::SingleTag,
305            topic_sys_flag: 1,
306            order: false,
307            attributes,
308        };
309        let encoded = config.encode();
310        assert!(encoded.contains("test_topic 8 8 4 SINGLE_TAG {\"key\":\"value\"}"));
311    }
312
313    #[test]
314    fn encode_topic_config_escapes_attribute_json_without_panicking() {
315        let mut attributes = HashMap::new();
316        attributes.insert(CheetahString::from("k\"ey"), CheetahString::from("v\\alue"));
317        let config = TopicConfig {
318            topic_name: Some(CheetahString::from("test_topic")),
319            read_queue_nums: 8,
320            write_queue_nums: 8,
321            perm: PermName::PERM_READ,
322            topic_filter_type: TopicFilterType::SingleTag,
323            topic_sys_flag: 1,
324            order: false,
325            attributes,
326        };
327
328        let encoded = config.encode();
329
330        assert!(encoded.ends_with(r#"{"k\"ey":"v\\alue"}"#));
331    }
332
333    #[test]
334    fn decode_topic_config() {
335        let mut config = TopicConfig::default();
336        let input = "test_topic 8 8 2 SingleTag {\"key\":\"value\"}";
337        let result = config.decode(input);
338        assert!(result);
339        assert_eq!(config.topic_name, Some(CheetahString::from("test_topic")));
340        assert_eq!(config.read_queue_nums, 8);
341        assert_eq!(config.write_queue_nums, 8);
342        assert_eq!(config.perm, PermName::PERM_WRITE);
343        assert_eq!(config.topic_filter_type, TopicFilterType::SingleTag);
344        assert_eq!(
345            config.attributes.get(&CheetahString::from("key")),
346            Some(&CheetahString::from("value"))
347        );
348    }
349
350    #[test]
351    fn get_topic_message_type_normal() {
352        let config = TopicConfig::default();
353        assert_eq!(config.get_topic_message_type(), TopicMessageType::Normal);
354    }
355
356    #[test]
357    fn get_topic_message_type_from_attributes() {
358        let mut config = TopicConfig::default();
359        config.attributes.insert(
360            CheetahString::from(TopicAttributes::topic_message_type_attribute().name()),
361            CheetahString::from("LITE"),
362        );
363        assert_eq!(config.get_topic_message_type(), TopicMessageType::Lite);
364    }
365
366    #[test]
367    fn get_lite_topic_expiration_defaults_to_minus_one_for_non_lite_topic() {
368        let config = TopicConfig::default();
369
370        assert_eq!(config.get_lite_topic_expiration(), -1);
371    }
372
373    #[test]
374    fn get_lite_topic_expiration_reads_attribute_for_lite_topic() {
375        let mut config = TopicConfig::default();
376        config.attributes.insert(
377            CheetahString::from(TopicAttributes::topic_message_type_attribute().name()),
378            CheetahString::from("LITE"),
379        );
380        config.attributes.insert(
381            CheetahString::from(TopicAttributes::lite_topic_expiration_attribute().name()),
382            CheetahString::from("600"),
383        );
384
385        assert_eq!(config.get_lite_topic_expiration(), 600);
386    }
387
388    #[test]
389    fn set_lite_topic_expiration_only_applies_to_lite_topics() {
390        let mut normal_config = TopicConfig::default();
391        normal_config.set_lite_topic_expiration(600);
392        assert_eq!(normal_config.get_lite_topic_expiration(), -1);
393
394        let mut lite_config = TopicConfig::default();
395        lite_config.attributes.insert(
396            CheetahString::from(TopicAttributes::topic_message_type_attribute().name()),
397            CheetahString::from("LITE"),
398        );
399        lite_config.set_lite_topic_expiration(600);
400        assert_eq!(lite_config.get_lite_topic_expiration(), 600);
401    }
402}
rocketmq_common/common/config.rs

rocketmq_common/common/
config.rs
