rocketmq_protocol/protocol/
namespace_util.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;
16use rocketmq_model::common::mix_all;
17use rocketmq_model::common::topic::TopicValidator;
18
19pub struct NamespaceUtil;
20
21impl NamespaceUtil {
22    pub const DLQ_PREFIX_LENGTH: usize = mix_all::DLQ_GROUP_TOPIC_PREFIX.len();
23    pub const NAMESPACE_SEPARATOR: char = '%';
24    pub const RETRY_PREFIX_LENGTH: usize = mix_all::RETRY_GROUP_TOPIC_PREFIX.len();
25    pub const STRING_BLANK: &'static str = "";
26
27    pub fn without_namespace(resource_with_namespace: &str) -> String {
28        if resource_with_namespace.is_empty() || NamespaceUtil::is_system_resource(resource_with_namespace) {
29            return resource_with_namespace.to_string();
30        }
31
32        let mut string_builder = String::new();
33        if NamespaceUtil::is_retry_topic(resource_with_namespace) {
34            string_builder.push_str(mix_all::RETRY_GROUP_TOPIC_PREFIX);
35        }
36        if NamespaceUtil::is_dlq_topic(resource_with_namespace) {
37            string_builder.push_str(mix_all::DLQ_GROUP_TOPIC_PREFIX);
38        }
39
40        if let Some(index) =
41            NamespaceUtil::without_retry_and_dlq(resource_with_namespace).find(NamespaceUtil::NAMESPACE_SEPARATOR)
42        {
43            let resource_without_namespace =
44                &NamespaceUtil::without_retry_and_dlq(resource_with_namespace)[index + 1..];
45            return string_builder + resource_without_namespace;
46        }
47
48        resource_with_namespace.to_string()
49    }
50
51    pub fn without_namespace_with_namespace(resource_with_namespace: &str, namespace: &str) -> String {
52        if resource_with_namespace.is_empty() || namespace.is_empty() {
53            return resource_with_namespace.to_string();
54        }
55
56        let resource_without_retry_and_dlq = NamespaceUtil::without_retry_and_dlq(resource_with_namespace);
57        if resource_without_retry_and_dlq.starts_with(&format!("{}{}", namespace, NamespaceUtil::NAMESPACE_SEPARATOR)) {
58            return NamespaceUtil::without_namespace(resource_with_namespace);
59        }
60
61        resource_with_namespace.to_string()
62    }
63
64    pub fn wrap_namespace(
65        namespace: impl Into<CheetahString>,
66        resource_without_namespace: impl Into<CheetahString>,
67    ) -> CheetahString {
68        let namespace = namespace.into();
69        let resource_without_namespace = resource_without_namespace.into();
70
71        if namespace.is_empty() || resource_without_namespace.is_empty() {
72            return resource_without_namespace;
73        }
74
75        if NamespaceUtil::is_system_resource(resource_without_namespace.as_str())
76            || NamespaceUtil::is_already_with_namespace(resource_without_namespace.as_str(), namespace.as_str())
77        {
78            return resource_without_namespace;
79        }
80
81        let mut string_builder = String::new();
82
83        if NamespaceUtil::is_retry_topic(resource_without_namespace.as_str()) {
84            string_builder.push_str(mix_all::RETRY_GROUP_TOPIC_PREFIX);
85        }
86
87        if NamespaceUtil::is_dlq_topic(resource_without_namespace.as_str()) {
88            string_builder.push_str(mix_all::DLQ_GROUP_TOPIC_PREFIX);
89        }
90        let resource_without_retry_and_dlq = NamespaceUtil::without_retry_and_dlq(resource_without_namespace.as_str());
91        CheetahString::from_string(
92            string_builder
93                + namespace.as_str()
94                + &NamespaceUtil::NAMESPACE_SEPARATOR.to_string()
95                + resource_without_retry_and_dlq,
96        )
97    }
98
99    pub fn is_already_with_namespace(resource: &str, namespace: &str) -> bool {
100        if namespace.is_empty() || resource.is_empty() || NamespaceUtil::is_system_resource(resource) {
101            return false;
102        }
103
104        let resource_without_retry_and_dlq = NamespaceUtil::without_retry_and_dlq(resource);
105
106        resource_without_retry_and_dlq.starts_with(&format!("{}{}", namespace, NamespaceUtil::NAMESPACE_SEPARATOR))
107    }
108
109    pub fn wrap_namespace_and_retry(
110        namespace: impl Into<CheetahString>,
111        consumer_group: impl Into<CheetahString>,
112    ) -> Option<CheetahString> {
113        let consumer_group = consumer_group.into();
114        if consumer_group.is_empty() {
115            return None;
116        }
117
118        let namespace = namespace.into();
119        let wrapped = NamespaceUtil::wrap_namespace(namespace, consumer_group);
120        Some(CheetahString::from_string(
121            mix_all::RETRY_GROUP_TOPIC_PREFIX.to_string() + wrapped.as_str(),
122        ))
123    }
124
125    pub fn get_namespace_from_resource(resource: &str) -> String {
126        if resource.is_empty() || NamespaceUtil::is_system_resource(resource) {
127            return NamespaceUtil::STRING_BLANK.to_string();
128        }
129        let resource_without_retry_and_dlq = NamespaceUtil::without_retry_and_dlq(resource);
130        if let Some(index) = resource_without_retry_and_dlq.find(NamespaceUtil::NAMESPACE_SEPARATOR) {
131            return resource_without_retry_and_dlq[..index].to_string();
132        }
133
134        NamespaceUtil::STRING_BLANK.to_string()
135    }
136
137    fn without_retry_and_dlq(original_resource: &str) -> &str {
138        if original_resource.is_empty() {
139            return NamespaceUtil::STRING_BLANK;
140        }
141        if NamespaceUtil::is_retry_topic(original_resource) {
142            return &original_resource[NamespaceUtil::RETRY_PREFIX_LENGTH..];
143        }
144
145        if NamespaceUtil::is_dlq_topic(original_resource) {
146            return &original_resource[NamespaceUtil::DLQ_PREFIX_LENGTH..];
147        }
148
149        original_resource
150    }
151
152    fn is_system_resource(resource: &str) -> bool {
153        if resource.is_empty() {
154            return false;
155        }
156        TopicValidator::is_system_topic(resource) || mix_all::is_sys_consumer_group(resource)
157    }
158
159    #[inline]
160    pub fn is_retry_topic(resource: &str) -> bool {
161        !resource.is_empty() && resource.starts_with(mix_all::RETRY_GROUP_TOPIC_PREFIX)
162    }
163
164    #[inline]
165    pub fn is_dlq_topic(resource: &str) -> bool {
166        !resource.is_empty() && resource.starts_with(mix_all::DLQ_GROUP_TOPIC_PREFIX)
167    }
168}
169
170#[cfg(test)]
171mod tests {
172    use super::*;
173
174    #[test]
175    fn without_namespace_returns_original_when_empty() {
176        assert_eq!(NamespaceUtil::without_namespace(""), "");
177    }
178
179    #[test]
180    fn without_namespace_returns_original_when_system_resource() {
181        assert_eq!(NamespaceUtil::without_namespace("SYS_TOPIC"), "SYS_TOPIC");
182    }
183
184    #[test]
185    fn without_namespace_removes_namespace() {
186        assert_eq!(
187            NamespaceUtil::without_namespace("my_namespace%my_resource"),
188            "my_resource"
189        );
190    }
191
192    #[test]
193    fn without_namespace_with_namespace_returns_original_when_empty() {
194        assert_eq!(NamespaceUtil::without_namespace_with_namespace("", "my_namespace"), "");
195    }
196
197    #[test]
198    fn without_namespace_with_namespace_removes_namespace() {
199        assert_eq!(
200            NamespaceUtil::without_namespace_with_namespace("my_namespace%my_resource", "my_namespace"),
201            "my_resource"
202        );
203    }
204
205    #[test]
206    fn wrap_namespace_returns_original_when_empty() {
207        assert_eq!(NamespaceUtil::wrap_namespace("my_namespace", ""), "");
208    }
209
210    #[test]
211    fn wrap_namespace_adds_namespace() {
212        assert_eq!(
213            NamespaceUtil::wrap_namespace("my_namespace", "my_resource"),
214            "my_namespace%my_resource"
215        );
216    }
217
218    #[test]
219    fn is_already_with_namespace_returns_false_when_empty() {
220        assert!(!NamespaceUtil::is_already_with_namespace("", "my_namespace"));
221    }
222
223    #[test]
224    fn is_already_with_namespace_returns_true_when_with_namespace() {
225        assert!(NamespaceUtil::is_already_with_namespace(
226            "my_namespace%my_resource",
227            "my_namespace"
228        ));
229    }
230
231    #[test]
232    fn wrap_namespace_and_retry_returns_none_when_empty() {
233        assert_eq!(NamespaceUtil::wrap_namespace_and_retry("my_namespace", ""), None);
234    }
235
236    #[test]
237    fn wrap_namespace_and_retry_adds_namespace_and_retry() {
238        assert_eq!(
239            NamespaceUtil::wrap_namespace_and_retry("my_namespace", "my_group").map(|s| s.as_str().to_string()),
240            Some("%RETRY%my_namespace%my_group".to_string())
241        );
242    }
243
244    #[test]
245    fn get_namespace_from_resource_returns_blank_when_empty() {
246        assert_eq!(NamespaceUtil::get_namespace_from_resource(""), "");
247    }
248
249    #[test]
250    fn get_namespace_from_resource_returns_namespace() {
251        assert_eq!(
252            NamespaceUtil::get_namespace_from_resource("my_namespace%my_resource"),
253            "my_namespace"
254        );
255    }
256
257    #[test]
258    fn without_retry_and_dlq_returns_original_when_empty() {
259        assert_eq!(NamespaceUtil::without_retry_and_dlq(""), "");
260    }
261
262    #[test]
263    fn without_retry_and_dlq_removes_retry_and_dlq() {
264        assert_eq!(
265            NamespaceUtil::without_retry_and_dlq("RETRY_GROUP_TOPIC_PREFIXmy_resource"),
266            "RETRY_GROUP_TOPIC_PREFIXmy_resource"
267        );
268        assert_eq!(
269            NamespaceUtil::without_retry_and_dlq("DLQ_GROUP_TOPIC_PREFIXmy_resource"),
270            "DLQ_GROUP_TOPIC_PREFIXmy_resource"
271        );
272    }
273
274    #[test]
275    fn is_system_resource_returns_false_when_empty() {
276        assert!(!NamespaceUtil::is_system_resource(""));
277    }
278
279    #[test]
280    fn is_system_resource_returns_true_when_system_resource() {
281        assert!(NamespaceUtil::is_system_resource("CID_RMQ_SYS_"));
282        assert!(NamespaceUtil::is_system_resource("TBW102"));
283    }
284
285    #[test]
286    fn is_retry_topic_returns_false_when_empty() {
287        assert!(!NamespaceUtil::is_retry_topic(""));
288    }
289
290    #[test]
291    fn is_retry_topic_returns_true_when_retry_topic() {
292        assert!(!NamespaceUtil::is_retry_topic("RETRY_GROUP_TOPIC_PREFIXmy_topic"));
293        assert!(NamespaceUtil::is_retry_topic("%RETRY%RETRY_GROUP_TOPIC_PREFIXmy_topic"));
294    }
295
296    #[test]
297    fn is_dlq_topic_returns_false_when_empty() {
298        assert!(!NamespaceUtil::is_dlq_topic(""));
299    }
300
301    #[test]
302    fn is_dlq_topic_returns_true_when_dlq_topic() {
303        assert!(!NamespaceUtil::is_dlq_topic("DLQ_GROUP_TOPIC_PREFIXmy_topic"));
304        assert!(NamespaceUtil::is_dlq_topic("%DLQ%DLQ_GROUP_TOPIC_PREFIXmy_topic"));
305    }
306}
rocketmq_protocol/protocol/namespace_util.rs

rocketmq_protocol/protocol/
namespace_util.rs
