rocketmq_model/common/attribute/
topic_attributes.rs

1// Copyright 2026 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::sync::Arc;
18use std::sync::OnceLock;
19
20use cheetah_string::CheetahString;
21
22use crate::common::attribute::enum_attribute::EnumAttribute;
23use crate::common::attribute::long_range_attribute::LongRangeAttribute;
24use crate::common::attribute::topic_message_type::TopicMessageType;
25use crate::common::attribute::Attribute;
26
27/// Defines attributes and configurations for RocketMQ topics
28pub struct TopicAttributes;
29
30impl TopicAttributes {
31    /// Queue type attribute defining storage structure (BatchCQ or SimpleCQ)
32    pub fn queue_type_attribute() -> &'static EnumAttribute {
33        static INSTANCE: OnceLock<EnumAttribute> = OnceLock::new();
34        INSTANCE.get_or_init(|| {
35            let mut valid_values = HashSet::new();
36            valid_values.insert("BatchCQ".into());
37            valid_values.insert("SimpleCQ".into());
38
39            EnumAttribute::new("queue.type".into(), false, valid_values, "SimpleCQ".into())
40        })
41    }
42
43    /// Cleanup policy attribute defining how messages are cleaned up (DELETE or COMPACTION)
44    pub fn cleanup_policy_attribute() -> &'static EnumAttribute {
45        static INSTANCE: OnceLock<EnumAttribute> = OnceLock::new();
46        INSTANCE.get_or_init(|| {
47            let mut valid_values = HashSet::new();
48            valid_values.insert("DELETE".into());
49            valid_values.insert("COMPACTION".into());
50
51            EnumAttribute::new("cleanup.policy".into(), false, valid_values, "DELETE".into())
52        })
53    }
54
55    /// Message type attribute defining the type of messages stored in the topic
56    pub fn topic_message_type_attribute() -> &'static EnumAttribute {
57        static INSTANCE: OnceLock<EnumAttribute> = OnceLock::new();
58        INSTANCE.get_or_init(|| {
59            EnumAttribute::new(
60                "message.type".into(),
61                true,
62                TopicMessageType::topic_message_type_set()
63                    .into_iter()
64                    .map(|item| item.into())
65                    .collect(),
66                TopicMessageType::Normal.to_string().into(),
67            )
68        })
69    }
70
71    /// Reserve time attribute defining how long messages are kept
72    pub fn topic_reserve_time_attribute() -> &'static LongRangeAttribute {
73        static INSTANCE: OnceLock<LongRangeAttribute> = OnceLock::new();
74        INSTANCE.get_or_init(|| LongRangeAttribute::new("reserve.time".into(), true, -1, i64::MAX, -1))
75    }
76
77    /// Lite topic expiration attribute defining per-topic lite TTL
78    pub fn lite_topic_expiration_attribute() -> &'static LongRangeAttribute {
79        static INSTANCE: OnceLock<LongRangeAttribute> = OnceLock::new();
80        INSTANCE.get_or_init(|| LongRangeAttribute::new("lite.topic.expiration".into(), true, -1, i64::MAX, -1))
81    }
82
83    /// Returns all defined attributes in a HashMap
84    pub fn all() -> &'static HashMap<CheetahString, Arc<dyn Attribute>> {
85        static ALL: OnceLock<HashMap<CheetahString, Arc<dyn Attribute>>> = OnceLock::new();
86        ALL.get_or_init(|| {
87            let mut map = HashMap::new();
88
89            let queue_type = Self::queue_type_attribute();
90            let cleanup_policy = Self::cleanup_policy_attribute();
91            let message_type = Self::topic_message_type_attribute();
92            let reserve_time = Self::topic_reserve_time_attribute();
93            let lite_expiration = Self::lite_topic_expiration_attribute();
94
95            map.insert(
96                queue_type.name().clone(),
97                Arc::new(queue_type.clone()) as Arc<dyn Attribute>,
98            );
99            map.insert(
100                cleanup_policy.name().clone(),
101                Arc::new(cleanup_policy.clone()) as Arc<dyn Attribute>,
102            );
103            map.insert(
104                message_type.name().clone(),
105                Arc::new(message_type.clone()) as Arc<dyn Attribute>,
106            );
107            map.insert(
108                reserve_time.name().clone(),
109                Arc::new(reserve_time.clone()) as Arc<dyn Attribute>,
110            );
111            map.insert(
112                lite_expiration.name().clone(),
113                Arc::new(lite_expiration.clone()) as Arc<dyn Attribute>,
114            );
115
116            map
117        })
118    }
119}
120
121#[cfg(test)]
122mod tests {
123
124    use super::*;
125
126    #[test]
127    fn queue_type_attribute_default_value() {
128        let attribute = TopicAttributes::queue_type_attribute();
129        assert_eq!(attribute.default_value(), "SimpleCQ");
130    }
131
132    #[test]
133    fn queue_type_attribute_valid_values() {
134        let attribute = TopicAttributes::queue_type_attribute();
135        let valid_values: HashSet<CheetahString> = ["BatchCQ", "SimpleCQ"]
136            .iter()
137            .cloned()
138            .map(CheetahString::from)
139            .collect();
140        assert_eq!(attribute.universe(), &valid_values);
141    }
142
143    #[test]
144    fn cleanup_policy_attribute_default_value() {
145        let attribute = TopicAttributes::cleanup_policy_attribute();
146        assert_eq!(attribute.default_value(), "DELETE");
147    }
148
149    #[test]
150    fn cleanup_policy_attribute_valid_values() {
151        let attribute = TopicAttributes::cleanup_policy_attribute();
152        let valid_values: HashSet<CheetahString> = ["DELETE", "COMPACTION"]
153            .iter()
154            .cloned()
155            .map(CheetahString::from)
156            .collect();
157        assert_eq!(attribute.universe(), &valid_values);
158    }
159
160    #[test]
161    fn topic_message_type_attribute_default_value() {
162        let attribute = TopicAttributes::topic_message_type_attribute();
163        assert_eq!(attribute.default_value(), "NORMAL");
164    }
165
166    #[test]
167    fn topic_message_type_attribute_valid_values() {
168        let attribute = TopicAttributes::topic_message_type_attribute();
169        let valid_values: HashSet<CheetahString> = TopicMessageType::topic_message_type_set()
170            .into_iter()
171            .map(CheetahString::from)
172            .collect();
173        assert_eq!(attribute.universe(), &valid_values);
174    }
175
176    #[test]
177    fn topic_reserve_time_attribute_default_value() {
178        let attribute = TopicAttributes::topic_reserve_time_attribute();
179        assert_eq!(attribute.default_value(), -1);
180    }
181
182    #[test]
183    fn topic_reserve_time_attribute_valid_range() {
184        let attribute = TopicAttributes::topic_reserve_time_attribute();
185        assert_eq!(attribute.min(), -1);
186        assert_eq!(attribute.max(), i64::MAX);
187    }
188
189    #[test]
190    fn lite_topic_expiration_attribute_default_value() {
191        let attribute = TopicAttributes::lite_topic_expiration_attribute();
192        assert_eq!(attribute.default_value(), -1);
193    }
194
195    #[test]
196    fn lite_topic_expiration_attribute_valid_range() {
197        let attribute = TopicAttributes::lite_topic_expiration_attribute();
198        assert_eq!(attribute.min(), -1);
199        assert_eq!(attribute.max(), i64::MAX);
200    }
201
202    #[test]
203    fn all_attributes_contains_all_defined_attributes() {
204        let all_attributes = TopicAttributes::all();
205        assert!(all_attributes.contains_key("queue.type"));
206        assert!(all_attributes.contains_key("cleanup.policy"));
207        assert!(all_attributes.contains_key("message.type"));
208        assert!(all_attributes.contains_key("reserve.time"));
209        assert!(all_attributes.contains_key("lite.topic.expiration"));
210    }
211}
rocketmq_model/common/attribute/topic_attributes.rs

rocketmq_model/common/attribute/
topic_attributes.rs
