rocketmq_model/common/
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::sync::LazyLock;
16
17use cheetah_string::CheetahString;
18use dashmap::DashSet;
19
20pub const TOPIC_MAX_LENGTH: usize = 127;
21
22/// Pre-computed valid character bitmap for topic/group name validation
23/// Allows: 0-9, a-z, A-Z, %, -, _, |
24const VALID_CHAR_BIT_MAP: [bool; 128] = {
25    let mut map = [false; 128];
26    map['%' as usize] = true;
27    map['-' as usize] = true;
28    map['_' as usize] = true;
29    map['|' as usize] = true;
30
31    let mut i = b'0';
32    while i <= b'9' {
33        map[i as usize] = true;
34        i += 1;
35    }
36
37    let mut i = b'A';
38    while i <= b'Z' {
39        map[i as usize] = true;
40        i += 1;
41    }
42
43    let mut i = b'a';
44    while i <= b'z' {
45        map[i as usize] = true;
46        i += 1;
47    }
48    map
49};
50
51static SYSTEM_TOPIC_SET: LazyLock<DashSet<CheetahString>> = LazyLock::new(|| {
52    let set = DashSet::new();
53    set.insert(CheetahString::from_static_str(
54        TopicValidator::AUTO_CREATE_TOPIC_KEY_TOPIC,
55    ));
56    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_SCHEDULE_TOPIC));
57    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_BENCHMARK_TOPIC));
58    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_TRANS_HALF_TOPIC));
59    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_TRACE_TOPIC));
60    set.insert(CheetahString::from_static_str(
61        TopicValidator::RMQ_SYS_TRANS_OP_HALF_TOPIC,
62    ));
63    set.insert(CheetahString::from_static_str(
64        TopicValidator::RMQ_SYS_TRANS_CHECK_MAX_TIME_TOPIC,
65    ));
66    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_SELF_TEST_TOPIC));
67    set.insert(CheetahString::from_static_str(
68        TopicValidator::RMQ_SYS_OFFSET_MOVED_EVENT,
69    ));
70    set.insert(CheetahString::from_static_str(
71        TopicValidator::RMQ_SYS_ROCKSDB_OFFSET_TOPIC,
72    ));
73    set.insert(CheetahString::from_static_str(
74        TopicValidator::RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC,
75    ));
76    set.insert(CheetahString::from_static_str(
77        TopicValidator::RMQ_SYS_ROCKSDB_TRANS_OP_HALF_TOPIC,
78    ));
79    set
80});
81
82static NOT_ALLOWED_SEND_TOPIC_SET: LazyLock<DashSet<CheetahString>> = LazyLock::new(|| {
83    let set = DashSet::new();
84    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_SCHEDULE_TOPIC));
85    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_TRANS_HALF_TOPIC));
86    set.insert(CheetahString::from_static_str(
87        TopicValidator::RMQ_SYS_TRANS_OP_HALF_TOPIC,
88    ));
89    set.insert(CheetahString::from_static_str(
90        TopicValidator::RMQ_SYS_TRANS_CHECK_MAX_TIME_TOPIC,
91    ));
92    set.insert(CheetahString::from_static_str(TopicValidator::RMQ_SYS_SELF_TEST_TOPIC));
93    set.insert(CheetahString::from_static_str(
94        TopicValidator::RMQ_SYS_OFFSET_MOVED_EVENT,
95    ));
96    set.insert(CheetahString::from_static_str(
97        TopicValidator::RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC,
98    ));
99    set.insert(CheetahString::from_static_str(
100        TopicValidator::RMQ_SYS_ROCKSDB_TRANS_OP_HALF_TOPIC,
101    ));
102    set
103});
104
105pub struct TopicValidator;
106
107impl TopicValidator {
108    pub const AUTO_CREATE_TOPIC_KEY_TOPIC: &'static str = "TBW102";
109    pub const RMQ_SYS_SCHEDULE_TOPIC: &'static str = "SCHEDULE_TOPIC_XXXX";
110    pub const RMQ_SYS_BENCHMARK_TOPIC: &'static str = "BenchmarkTest";
111    pub const RMQ_SYS_TRANS_HALF_TOPIC: &'static str = "RMQ_SYS_TRANS_HALF_TOPIC";
112    pub const RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC: &'static str = "RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC";
113    pub const RMQ_SYS_TRACE_TOPIC: &'static str = "RMQ_SYS_TRACE_TOPIC";
114    pub const RMQ_SYS_TRANS_OP_HALF_TOPIC: &'static str = "RMQ_SYS_TRANS_OP_HALF_TOPIC";
115    pub const RMQ_SYS_ROCKSDB_TRANS_OP_HALF_TOPIC: &'static str = "RMQ_SYS_ROCKSDB_TRANS_OP_HALF_TOPIC";
116    pub const RMQ_SYS_TRANS_CHECK_MAX_TIME_TOPIC: &'static str = "TRANS_CHECK_MAX_TIME_TOPIC";
117    pub const RMQ_SYS_SELF_TEST_TOPIC: &'static str = "SELF_TEST_TOPIC";
118    pub const RMQ_SYS_OFFSET_MOVED_EVENT: &'static str = "OFFSET_MOVED_EVENT";
119    pub const RMQ_SYS_ROCKSDB_OFFSET_TOPIC: &'static str = "CHECKPOINT_TOPIC";
120
121    pub const SYSTEM_TOPIC_PREFIX: &'static str = "rmq_sys_";
122    pub const SYNC_BROKER_MEMBER_GROUP_PREFIX: &'static str = "rmq_sys_SYNC_BROKER_MEMBER_";
123}
124
125impl TopicValidator {
126    /// Check if topic or group name contains illegal characters
127    ///
128    /// # Performance
129    /// Uses pre-computed bitmap for O(1) character validation
130    #[inline]
131    pub fn is_topic_or_group_illegal(name: &str) -> bool {
132        name.bytes()
133            .any(|b| (b as usize) >= VALID_CHAR_BIT_MAP.len() || !VALID_CHAR_BIT_MAP[b as usize])
134    }
135
136    /// Validate topic name according to RocketMQ rules
137    ///
138    /// # Rules
139    /// - Cannot be blank
140    /// - Must contain only: 0-9, a-z, A-Z, %, -, _, |
141    /// - Length must not exceed TOPIC_MAX_LENGTH (127)
142    #[inline]
143    pub fn validate_topic(topic: &str) -> ValidateTopicResult {
144        // Fast path: check length first (cheaper than trim)
145        if topic.is_empty() || topic.len() > TOPIC_MAX_LENGTH {
146            if topic.is_empty() || topic.trim().is_empty() {
147                const REMARK: &str = "The specified topic is blank.";
148                return ValidateTopicResult {
149                    valid: false,
150                    remark: CheetahString::from_static_str(REMARK),
151                };
152            }
153            return ValidateTopicResult {
154                valid: false,
155                remark: CheetahString::from(format!(
156                    "The specified topic is longer than topic max length {TOPIC_MAX_LENGTH}."
157                )),
158            };
159        }
160
161        if Self::is_topic_or_group_illegal(topic) {
162            const REMARK: &str = "The specified topic contains illegal characters, allowing only ^[%|a-zA-Z0-9_-]+$";
163            return ValidateTopicResult {
164                valid: false,
165                remark: CheetahString::from_static_str(REMARK),
166            };
167        }
168
169        ValidateTopicResult {
170            valid: true,
171            remark: CheetahString::empty(),
172        }
173    }
174
175    #[inline]
176    pub fn is_system_topic(topic: &str) -> bool {
177        // Fast path: check prefix first (no lock needed)
178        if topic.starts_with(TopicValidator::SYSTEM_TOPIC_PREFIX) {
179            return true;
180        }
181        // DashSet::contains is lock-free for reads
182        SYSTEM_TOPIC_SET.iter().any(|entry| entry.as_str() == topic)
183    }
184
185    #[inline]
186    pub fn is_not_allowed_send_topic(topic: &str) -> bool {
187        // DashSet iteration is lock-free
188        NOT_ALLOWED_SEND_TOPIC_SET.iter().any(|entry| entry.as_str() == topic)
189    }
190
191    pub fn add_system_topic(system_topic: impl Into<CheetahString>) {
192        SYSTEM_TOPIC_SET.insert(system_topic.into());
193    }
194
195    pub fn get_system_topic_set() -> Vec<CheetahString> {
196        SYSTEM_TOPIC_SET.iter().map(|entry| entry.clone()).collect()
197    }
198
199    pub fn get_not_allowed_send_topic_set() -> Vec<CheetahString> {
200        NOT_ALLOWED_SEND_TOPIC_SET.iter().map(|entry| entry.clone()).collect()
201    }
202}
203
204pub struct ValidateTopicResult {
205    valid: bool,
206    remark: CheetahString,
207}
208
209impl ValidateTopicResult {
210    #[inline]
211    pub fn valid(&self) -> bool {
212        self.valid
213    }
214
215    #[inline]
216    pub fn remark(&self) -> &CheetahString {
217        &self.remark
218    }
219
220    #[inline]
221    pub fn take_remark(self) -> CheetahString {
222        self.remark
223    }
224
225    #[inline]
226    pub fn set_valid(&mut self, valid: bool) {
227        self.valid = valid;
228    }
229
230    #[inline]
231    pub fn set_remark(&mut self, remark: CheetahString) {
232        self.remark = remark;
233    }
234}
235
236#[cfg(test)]
237mod tests {
238    use super::*;
239    #[test]
240    fn validate_topic_with_valid_topic() {
241        let result = TopicValidator::validate_topic("valid_topic");
242        assert!(result.valid());
243        assert_eq!(result.remark(), "");
244    }
245
246    #[test]
247    fn validate_topic_with_empty_topic() {
248        let result = TopicValidator::validate_topic("");
249        assert!(!result.valid());
250        assert_eq!(result.remark(), "The specified topic is blank.");
251    }
252
253    #[test]
254    fn validate_topic_with_illegal_characters() {
255        let result = TopicValidator::validate_topic("invalid@topic");
256        assert!(!result.valid());
257        assert_eq!(
258            result.remark(),
259            "The specified topic contains illegal characters, allowing only ^[%|a-zA-Z0-9_-]+$"
260        );
261    }
262
263    #[test]
264    fn validate_topic_with_exceeding_length() {
265        let long_topic = "a".repeat(TOPIC_MAX_LENGTH + 1);
266        let result = TopicValidator::validate_topic(&long_topic);
267        assert!(!result.valid());
268        assert_eq!(
269            result.remark().as_str(),
270            format!(
271                "The specified topic is longer than topic max length {}.",
272                TOPIC_MAX_LENGTH
273            )
274        );
275    }
276
277    #[test]
278    fn is_system_topic_with_system_topic() {
279        assert!(TopicValidator::is_system_topic(TopicValidator::RMQ_SYS_SCHEDULE_TOPIC));
280        assert!(TopicValidator::is_system_topic(
281            TopicValidator::RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC
282        ));
283        assert!(TopicValidator::is_system_topic(
284            TopicValidator::RMQ_SYS_ROCKSDB_TRANS_OP_HALF_TOPIC
285        ));
286    }
287
288    #[test]
289    fn is_system_topic_with_non_system_topic() {
290        assert!(!TopicValidator::is_system_topic("non_system_topic"));
291    }
292
293    #[test]
294    fn is_not_allowed_send_topic_with_not_allowed_topic() {
295        assert!(TopicValidator::is_not_allowed_send_topic(
296            TopicValidator::RMQ_SYS_SCHEDULE_TOPIC
297        ));
298        assert!(TopicValidator::is_not_allowed_send_topic(
299            TopicValidator::RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC
300        ));
301        assert!(TopicValidator::is_not_allowed_send_topic(
302            TopicValidator::RMQ_SYS_ROCKSDB_TRANS_OP_HALF_TOPIC
303        ));
304    }
305
306    #[test]
307    fn is_not_allowed_send_topic_with_allowed_topic() {
308        assert!(!TopicValidator::is_not_allowed_send_topic("allowed_topic"));
309    }
310
311    #[test]
312    fn add_system_topic_adds_new_topic() {
313        TopicValidator::add_system_topic("new_system_topic");
314        assert!(TopicValidator::is_system_topic("new_system_topic"));
315    }
316
317    #[test]
318    fn get_system_topic_set_returns_all_system_topics() {
319        let system_topics = TopicValidator::get_system_topic_set();
320        assert!(system_topics
321            .iter()
322            .any(|s| s.as_str() == TopicValidator::RMQ_SYS_SCHEDULE_TOPIC));
323    }
324
325    #[test]
326    fn get_not_allowed_send_topic_set_returns_all_not_allowed_topics() {
327        let not_allowed_topics = TopicValidator::get_not_allowed_send_topic_set();
328        assert!(not_allowed_topics
329            .iter()
330            .any(|s| s.as_str() == TopicValidator::RMQ_SYS_SCHEDULE_TOPIC));
331    }
332}
rocketmq_model/common/topic.rs

rocketmq_model/common/
topic.rs
