rocketmq_model/
topic.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::collections::HashSet;
17use std::fmt;
18use std::fmt::Display;
19
20use cheetah_string::CheetahString;
21use serde::Deserialize;
22use serde::Deserializer;
23use serde::Serialize;
24use serde::Serializer;
25
26const DEFAULT_TOPIC_PERMISSION: u32 = 0x1 << 2 | 0x1 << 1;
27const MESSAGE_TYPE_ATTRIBUTE: &str = "message.type";
28const LITE_EXPIRATION_ATTRIBUTE: &str = "lite.topic.expiration";
29
30pub const RETRY_GROUP_TOPIC_PREFIX: &str = "%RETRY%";
31pub const DLQ_GROUP_TOPIC_PREFIX: &str = "%DLQ%";
32pub const SYSTEM_TOPIC_PREFIX: &str = "rmq_sys_";
33pub const SYSTEM_TOPICS: &[&str] = &[
34    "SCHEDULE_TOPIC_XXXX",
35    "BenchmarkTest",
36    "RMQ_SYS_TRANS_HALF_TOPIC",
37    "RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC",
38    "RMQ_SYS_TRACE_TOPIC",
39    "RMQ_SYS_TRANS_OP_HALF_TOPIC",
40    "RMQ_SYS_ROCKSDB_TRANS_OP_HALF_TOPIC",
41    "TRANS_CHECK_MAX_TIME_TOPIC",
42    "SELF_TEST_TOPIC",
43    "OFFSET_MOVED_EVENT",
44    "CHECKPOINT_TOPIC",
45];
46
47/// Returns whether a topic belongs to the stable RocketMQ system-topic set.
48pub fn is_system_topic(topic: &str) -> bool {
49    topic.starts_with(SYSTEM_TOPIC_PREFIX) || SYSTEM_TOPICS.contains(&topic)
50}
51
52/// Topic filter expression mode.
53#[derive(Clone, Default, Eq, PartialEq, Copy)]
54pub enum TopicFilterType {
55    #[default]
56    SingleTag,
57    MultiTag,
58}
59
60impl TopicFilterType {
61    pub const fn as_str(&self) -> &'static str {
62        match self {
63            Self::SingleTag => "SINGLE_TAG",
64            Self::MultiTag => "MULTI_TAG",
65        }
66    }
67}
68
69impl Display for TopicFilterType {
70    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
71        f.write_str(self.as_str())
72    }
73}
74
75impl fmt::Debug for TopicFilterType {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        f.write_str(self.as_str())
78    }
79}
80
81impl From<&str> for TopicFilterType {
82    fn from(value: &str) -> Self {
83        match value {
84            "MULTI_TAG" => Self::MultiTag,
85            _ => Self::SingleTag,
86        }
87    }
88}
89
90impl From<String> for TopicFilterType {
91    fn from(value: String) -> Self {
92        Self::from(value.as_str())
93    }
94}
95
96impl From<i32> for TopicFilterType {
97    fn from(value: i32) -> Self {
98        match value {
99            1 => Self::MultiTag,
100            _ => Self::SingleTag,
101        }
102    }
103}
104
105impl Serialize for TopicFilterType {
106    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
107    where
108        S: Serializer,
109    {
110        serializer.serialize_str(self.as_str())
111    }
112}
113
114impl<'de> Deserialize<'de> for TopicFilterType {
115    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
116    where
117        D: Deserializer<'de>,
118    {
119        let value = String::deserialize(deserializer)?;
120        match value.as_str() {
121            "SINGLE_TAG" => Ok(Self::SingleTag),
122            "MULTI_TAG" => Ok(Self::MultiTag),
123            _ => Err(serde::de::Error::unknown_variant(&value, &["SINGLE_TAG", "MULTI_TAG"])),
124        }
125    }
126}
127
128/// Semantic message kind associated with a topic.
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
130pub enum TopicMessageType {
131    Unspecified,
132    Normal,
133    Fifo,
134    Delay,
135    Transaction,
136    Priority,
137    Lite,
138    Mixed,
139}
140
141impl From<String> for TopicMessageType {
142    fn from(value: String) -> Self {
143        match value.to_uppercase().as_str() {
144            "UNSPECIFIED" => Self::Unspecified,
145            "NORMAL" => Self::Normal,
146            "FIFO" => Self::Fifo,
147            "DELAY" => Self::Delay,
148            "TRANSACTION" => Self::Transaction,
149            "PRIORITY" => Self::Priority,
150            "LITE" => Self::Lite,
151            "MIXED" => Self::Mixed,
152            _ => Self::Unspecified,
153        }
154    }
155}
156
157impl TopicMessageType {
158    pub const fn as_str(&self) -> &'static str {
159        match self {
160            Self::Unspecified => "UNSPECIFIED",
161            Self::Normal => "NORMAL",
162            Self::Fifo => "FIFO",
163            Self::Delay => "DELAY",
164            Self::Transaction => "TRANSACTION",
165            Self::Priority => "PRIORITY",
166            Self::Lite => "LITE",
167            Self::Mixed => "MIXED",
168        }
169    }
170
171    pub fn topic_message_type_set() -> HashSet<String> {
172        [
173            Self::Unspecified,
174            Self::Normal,
175            Self::Fifo,
176            Self::Delay,
177            Self::Transaction,
178            Self::Priority,
179            Self::Lite,
180            Self::Mixed,
181        ]
182        .into_iter()
183        .map(|value| value.as_str().to_owned())
184        .collect()
185    }
186
187    pub fn parse_from_message_property(properties: &HashMap<String, String>) -> Self {
188        if properties
189            .get("TRAN_MSG")
190            .is_some_and(|value| value.eq_ignore_ascii_case("true"))
191        {
192            Self::Transaction
193        } else if ["DELAY", "TIMER_DELIVER_MS", "TIMER_DELAY_SEC", "TIMER_DELAY_MS"]
194            .iter()
195            .any(|key| properties.contains_key(*key))
196        {
197            Self::Delay
198        } else if properties.contains_key("__SHARDINGKEY") {
199            Self::Fifo
200        } else if properties.contains_key("_SYS_MSG_PRIORITY_") {
201            Self::Priority
202        } else if properties.contains_key("__LITE_TOPIC") {
203            Self::Lite
204        } else {
205            Self::Normal
206        }
207    }
208
209    pub fn get_metrics_value(&self) -> String {
210        self.as_str().to_ascii_lowercase()
211    }
212}
213
214impl Display for TopicMessageType {
215    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
216        f.write_str(self.as_str())
217    }
218}
219
220/// Persistent topic configuration value object.
221#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
222#[serde(rename_all = "camelCase")]
223pub struct TopicConfig {
224    pub topic_name: Option<CheetahString>,
225    pub read_queue_nums: u32,
226    pub write_queue_nums: u32,
227    pub perm: u32,
228    pub topic_filter_type: TopicFilterType,
229    pub topic_sys_flag: u32,
230    pub order: bool,
231    pub attributes: HashMap<CheetahString, CheetahString>,
232}
233
234impl TopicConfig {
235    const DEFAULT_READ_QUEUE_NUMS: u32 = 16;
236    const DEFAULT_WRITE_QUEUE_NUMS: u32 = 16;
237    const SEPARATOR: &'static str = " ";
238
239    pub fn new(topic_name: impl Into<CheetahString>) -> Self {
240        Self {
241            topic_name: Some(topic_name.into()),
242            ..Self::default()
243        }
244    }
245
246    pub fn with_queues(topic_name: impl Into<CheetahString>, read_queue_nums: u32, write_queue_nums: u32) -> Self {
247        Self {
248            read_queue_nums,
249            write_queue_nums,
250            ..Self::new(topic_name)
251        }
252    }
253
254    pub fn with_perm(
255        topic_name: impl Into<CheetahString>,
256        read_queue_nums: u32,
257        write_queue_nums: u32,
258        perm: u32,
259    ) -> Self {
260        Self {
261            read_queue_nums,
262            write_queue_nums,
263            perm,
264            ..Self::new(topic_name)
265        }
266    }
267
268    pub fn with_sys_flag(
269        topic_name: impl Into<CheetahString>,
270        read_queue_nums: u32,
271        write_queue_nums: u32,
272        perm: u32,
273        topic_sys_flag: u32,
274    ) -> Self {
275        Self {
276            read_queue_nums,
277            write_queue_nums,
278            perm,
279            topic_sys_flag,
280            ..Self::new(topic_name)
281        }
282    }
283
284    pub fn get_topic_message_type(&self) -> TopicMessageType {
285        self.attributes
286            .get(MESSAGE_TYPE_ATTRIBUTE)
287            .map(|value| TopicMessageType::from(value.to_string()))
288            .unwrap_or(TopicMessageType::Normal)
289    }
290
291    pub fn set_lite_topic_expiration(&mut self, lite_topic_expiration: i32) {
292        if self.get_topic_message_type() == TopicMessageType::Lite {
293            self.attributes.insert(
294                CheetahString::from_static_str(LITE_EXPIRATION_ATTRIBUTE),
295                CheetahString::from_string(lite_topic_expiration.to_string()),
296            );
297        }
298    }
299
300    pub fn get_lite_topic_expiration(&self) -> i32 {
301        if self.get_topic_message_type() != TopicMessageType::Lite {
302            return -1;
303        }
304        self.attributes
305            .get(LITE_EXPIRATION_ATTRIBUTE)
306            .and_then(|value| value.parse().ok())
307            .unwrap_or(-1)
308    }
309
310    pub fn encode(&self) -> String {
311        let mut parts = vec![
312            self.topic_name.clone().unwrap_or_default().to_string(),
313            self.read_queue_nums.to_string(),
314            self.write_queue_nums.to_string(),
315            self.perm.to_string(),
316            self.topic_filter_type.to_string(),
317        ];
318        if !self.attributes.is_empty() {
319            parts.push(serde_json::to_string(&self.attributes).unwrap_or_default());
320        }
321        parts.join(Self::SEPARATOR)
322    }
323
324    pub fn decode(&mut self, input: &str) -> bool {
325        let parts: Vec<&str> = input.splitn(6, Self::SEPARATOR).collect();
326        if parts.len() < 5 {
327            return false;
328        }
329        self.topic_name = Some(parts[0].into());
330        self.read_queue_nums = parts[1].parse().unwrap_or(Self::DEFAULT_READ_QUEUE_NUMS);
331        self.write_queue_nums = parts[2].parse().unwrap_or(Self::DEFAULT_WRITE_QUEUE_NUMS);
332        self.perm = parts[3].parse().unwrap_or(DEFAULT_TOPIC_PERMISSION);
333        self.topic_filter_type = parts[4].into();
334        if let Some(attributes) = parts.get(5).and_then(|value| serde_json::from_str(value).ok()) {
335            self.attributes = attributes;
336        }
337        true
338    }
339
340    pub fn get_read_queue_nums(&self) -> u32 {
341        self.read_queue_nums
342    }
343}
344
345impl Default for TopicConfig {
346    fn default() -> Self {
347        Self {
348            topic_name: None,
349            read_queue_nums: Self::DEFAULT_READ_QUEUE_NUMS,
350            write_queue_nums: Self::DEFAULT_WRITE_QUEUE_NUMS,
351            perm: DEFAULT_TOPIC_PERMISSION,
352            topic_filter_type: TopicFilterType::SingleTag,
353            topic_sys_flag: 0,
354            order: false,
355            attributes: HashMap::new(),
356        }
357    }
358}
359
360impl Display for TopicConfig {
361    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
362        write!(
363            f,
364            "TopicConfig {{ topic_name: {:?}, read_queue_nums: {}, write_queue_nums: {}, perm: {}, topic_filter_type: \
365             {}, topic_sys_flag: {}, order: {}, attributes: {:?} }}",
366            self.topic_name,
367            self.read_queue_nums,
368            self.write_queue_nums,
369            self.perm,
370            self.topic_filter_type,
371            self.topic_sys_flag,
372            self.order,
373            self.attributes
374        )
375    }
376}
377
378#[cfg(test)]
379mod topic_message_type_tests {
380    use std::collections::HashMap;
381
382    use super::*;
383
384    /// A precedence case: a description, the marker properties to classify, and the
385    /// expected message type.
386    type PrecedenceCase = (&'static str, &'static [(&'static str, &'static str)], TopicMessageType);
387
388    fn props(pairs: &[(&str, &str)]) -> HashMap<String, String> {
389        pairs
390            .iter()
391            .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
392            .collect()
393    }
394
395    #[test]
396    fn parse_from_message_property_marker_precedence() {
397        // Each row carries every lower-precedence marker alongside the winning one, so it
398        // asserts which marker wins a tie rather than merely that a lone marker maps through.
399        let cases: &[PrecedenceCase] = &[
400            (
401                "TRAN_MSG=true outranks delay and sharding markers",
402                &[("TRAN_MSG", "true"), ("DELAY", "1"), ("__SHARDINGKEY", "k")],
403                TopicMessageType::Transaction,
404            ),
405            (
406                "a delay marker outranks sharding/priority/lite; TRAN_MSG=false does not force Transaction",
407                &[
408                    ("TRAN_MSG", "false"),
409                    ("DELAY", "3"),
410                    ("__SHARDINGKEY", "k"),
411                    ("_SYS_MSG_PRIORITY_", "5"),
412                    ("__LITE_TOPIC", "1"),
413                ],
414                TopicMessageType::Delay,
415            ),
416            (
417                "__SHARDINGKEY outranks priority and lite",
418                &[
419                    ("__SHARDINGKEY", "k"),
420                    ("_SYS_MSG_PRIORITY_", "5"),
421                    ("__LITE_TOPIC", "1"),
422                ],
423                TopicMessageType::Fifo,
424            ),
425            (
426                "_SYS_MSG_PRIORITY_ outranks lite",
427                &[("_SYS_MSG_PRIORITY_", "5"), ("__LITE_TOPIC", "1")],
428                TopicMessageType::Priority,
429            ),
430            (
431                "__LITE_TOPIC alone selects Lite",
432                &[("__LITE_TOPIC", "1")],
433                TopicMessageType::Lite,
434            ),
435            (
436                "TRAN_MSG=false with no other marker selects Normal",
437                &[("TRAN_MSG", "false")],
438                TopicMessageType::Normal,
439            ),
440            ("an empty property map selects Normal", &[], TopicMessageType::Normal),
441        ];
442
443        for (desc, pairs, expected) in cases {
444            let got = TopicMessageType::parse_from_message_property(&props(pairs));
445            assert_eq!(got, *expected, "{desc}");
446        }
447    }
448
449    #[test]
450    fn parse_from_message_property_treats_every_timer_key_as_delay() {
451        for key in ["DELAY", "TIMER_DELIVER_MS", "TIMER_DELAY_SEC", "TIMER_DELAY_MS"] {
452            let got = TopicMessageType::parse_from_message_property(&props(&[(key, "1")]));
453            assert_eq!(got, TopicMessageType::Delay, "{key} should select Delay");
454        }
455    }
456
457    #[test]
458    fn parse_from_message_property_tran_msg_true_is_case_insensitive() {
459        for value in ["true", "TRUE", "TrUe"] {
460            let got = TopicMessageType::parse_from_message_property(&props(&[("TRAN_MSG", value)]));
461            assert_eq!(
462                got,
463                TopicMessageType::Transaction,
464                "TRAN_MSG={value} should select Transaction"
465            );
466        }
467        // A TRAN_MSG value other than a case-insensitive "true" must not be read as Transaction.
468        let got = TopicMessageType::parse_from_message_property(&props(&[("TRAN_MSG", "1")]));
469        assert_eq!(got, TopicMessageType::Normal, "TRAN_MSG=1 is not Transaction");
470    }
471}
rocketmq_model/topic.rs

rocketmq_model/
topic.rs
