rocketmq_model/common/
key_builder.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 crate::topic::RETRY_GROUP_TOPIC_PREFIX;
16
17use super::pop_retry_policy::PopRetryTopicVersion;
18
19pub const POP_ORDER_REVIVE_QUEUE: i32 = 999;
20pub const POP_RETRY_SEPARATOR_V1: char = '_';
21pub const POP_RETRY_SEPARATOR_V2: char = '+';
22pub const POP_RETRY_REGEX_SEPARATOR_V2: &str = "\\+";
23
24pub struct KeyBuilder;
25
26impl KeyBuilder {
27    pub fn build_pop_retry_topic(topic: &str, cid: &str, enable_retry_v2: bool) -> String {
28        let version = if enable_retry_v2 {
29            PopRetryTopicVersion::V2
30        } else {
31            PopRetryTopicVersion::V1
32        };
33        Self::build_pop_retry_topic_for_version(topic, cid, version)
34    }
35
36    pub fn build_pop_retry_topic_for_version(topic: &str, cid: &str, version: PopRetryTopicVersion) -> String {
37        match version {
38            PopRetryTopicVersion::V1 => Self::build_pop_retry_topic_v1(topic, cid),
39            PopRetryTopicVersion::V2 => Self::build_pop_retry_topic_v2(topic, cid),
40        }
41    }
42
43    pub fn build_pop_retry_topic_default(topic: &str, cid: &str) -> String {
44        format!("{RETRY_GROUP_TOPIC_PREFIX}{cid}{POP_RETRY_SEPARATOR_V1}{topic}")
45    }
46
47    pub fn build_pop_retry_topic_v2(topic: &str, cid: &str) -> String {
48        format!("{RETRY_GROUP_TOPIC_PREFIX}{cid}{POP_RETRY_SEPARATOR_V2}{topic}")
49    }
50
51    pub fn build_pop_retry_topic_v1(topic: &str, cid: &str) -> String {
52        format!("{RETRY_GROUP_TOPIC_PREFIX}{cid}{POP_RETRY_SEPARATOR_V1}{topic}")
53    }
54
55    pub fn parse_normal_topic(topic: &str, cid: &str) -> String {
56        if topic.starts_with(RETRY_GROUP_TOPIC_PREFIX) {
57            if topic.starts_with(&format!("{RETRY_GROUP_TOPIC_PREFIX}{cid}{POP_RETRY_SEPARATOR_V2}")) {
58                return topic[RETRY_GROUP_TOPIC_PREFIX.len() + cid.len() + 1..].to_string();
59            }
60            return topic[RETRY_GROUP_TOPIC_PREFIX.len() + cid.len() + 1..].to_string();
61        }
62        topic.to_string()
63    }
64
65    pub fn parse_normal_topic_default(retry_topic: &str) -> String {
66        if KeyBuilder::is_pop_retry_topic_v2(retry_topic) {
67            let result: Vec<&str> = retry_topic.split(POP_RETRY_SEPARATOR_V2).collect();
68            if result.len() == 2 {
69                return result[1].to_string();
70            }
71        }
72        retry_topic.to_string()
73    }
74
75    pub fn parse_group(retry_topic: &str) -> String {
76        if KeyBuilder::is_pop_retry_topic_v2(retry_topic) {
77            let result: Vec<&str> = retry_topic.split(POP_RETRY_SEPARATOR_V2).collect();
78            if result.len() == 2 {
79                return result[0][RETRY_GROUP_TOPIC_PREFIX.len()..].to_string();
80            }
81        }
82        retry_topic[RETRY_GROUP_TOPIC_PREFIX.len()..].to_string()
83    }
84
85    pub fn build_polling_key(topic: &str, cid: &str, queue_id: i32) -> String {
86        format!("{}{}{}{}{}", topic, "@", cid, "@", queue_id)
87    }
88
89    pub fn is_pop_retry_topic_v2(retry_topic: &str) -> bool {
90        retry_topic.starts_with(RETRY_GROUP_TOPIC_PREFIX) && retry_topic.contains(POP_RETRY_SEPARATOR_V2)
91    }
92
93    pub fn classify_pop_retry_topic(retry_topic: &str) -> Option<PopRetryTopicVersion> {
94        if Self::is_pop_retry_topic_v2(retry_topic) {
95            Some(PopRetryTopicVersion::V2)
96        } else if retry_topic.starts_with(RETRY_GROUP_TOPIC_PREFIX)
97            && retry_topic[RETRY_GROUP_TOPIC_PREFIX.len()..].contains(POP_RETRY_SEPARATOR_V1)
98        {
99            Some(PopRetryTopicVersion::V1)
100        } else {
101            None
102        }
103    }
104
105    pub fn parse_pop_retry_topic<'a>(retry_topic: &'a str, cid: &str) -> Option<(PopRetryTopicVersion, &'a str)> {
106        let prefix = format!("{RETRY_GROUP_TOPIC_PREFIX}{cid}");
107        let suffix = retry_topic.strip_prefix(&prefix)?;
108        if let Some(topic) = suffix.strip_prefix(POP_RETRY_SEPARATOR_V2) {
109            (!topic.is_empty()).then_some((PopRetryTopicVersion::V2, topic))
110        } else if let Some(topic) = suffix.strip_prefix(POP_RETRY_SEPARATOR_V1) {
111            (!topic.is_empty()).then_some((PopRetryTopicVersion::V1, topic))
112        } else {
113            None
114        }
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use super::KeyBuilder;
121
122    #[test]
123    fn pop_retry_v2_builder_round_trips() {
124        let retry = KeyBuilder::build_pop_retry_topic_v2("orders", "workers");
125        assert_eq!(retry, "%RETRY%workers+orders");
126        assert_eq!(KeyBuilder::parse_normal_topic_default(&retry), "orders");
127        assert_eq!(KeyBuilder::parse_group(&retry), "workers");
128    }
129
130    #[test]
131    fn literal_v2_topic_is_parsed() {
132        assert_eq!(
133            KeyBuilder::parse_normal_topic_default("%RETRY%workers+orders"),
134            "orders"
135        );
136        assert_eq!(KeyBuilder::parse_group("%RETRY%workers+orders"), "workers");
137    }
138
139    #[test]
140    fn non_v2_and_multiple_separator_fallbacks_are_preserved() {
141        for topic in ["orders", "%RETRY%workers_orders", "%RETRY%workers+orders+extra"] {
142            assert_eq!(KeyBuilder::parse_normal_topic_default(topic), topic);
143        }
144        assert_eq!(KeyBuilder::parse_group("%RETRY%workers"), "workers");
145        assert_eq!(KeyBuilder::parse_group("%RETRY%workers_orders"), "workers_orders");
146        assert_eq!(
147            KeyBuilder::parse_group("%RETRY%workers+orders+extra"),
148            "workers+orders+extra"
149        );
150    }
151}
rocketmq_model/common/key_builder.rs

rocketmq_model/common/
key_builder.rs
