rocketmq_common/common/
message.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::any::Any;
16use std::collections::HashMap;
17use std::collections::HashSet;
18use std::fmt;
19use std::fmt::Debug;
20use std::fmt::Display;
21use std::string::ToString;
22use std::sync::LazyLock;
23
24use bytes::Buf;
25use bytes::Bytes;
26use cheetah_string::CheetahString;
27use rocketmq_error::RocketMQError;
28use rocketmq_error::RocketMQResult;
29
30pub mod message_accessor;
31pub mod message_batch;
32pub mod message_batch_v2;
33pub mod message_body;
34pub mod message_builder;
35pub mod message_client_ext;
36pub mod message_client_id_setter;
37pub mod message_decoder;
38pub mod message_enum;
39pub mod message_ext;
40pub mod message_ext_broker_inner;
41pub mod message_flag;
42pub mod message_id;
43pub mod message_property;
44pub mod message_queue;
45pub mod message_queue_assignment;
46pub mod message_queue_for_c;
47pub mod message_single;
48
49// New refactored message types
50pub mod broker_message;
51pub mod message_envelope;
52pub mod routing_context;
53pub mod storage_metadata;
54
55/// Defines the interface for RocketMQ message operations.
56///
57/// Provides methods for managing message properties, keys, tags, body, and metadata.
58pub trait MessageTrait: Any + Display + Debug {
59    /// Sets the keys for the message.
60    #[inline]
61    fn set_keys(&mut self, keys: CheetahString) {
62        self.put_property(CheetahString::from_static_str(MessageConst::PROPERTY_KEYS), keys);
63    }
64
65    /// Adds a property to the message.
66    fn put_property(&mut self, key: CheetahString, value: CheetahString);
67
68    /// Removes the specified property from the message.
69    fn clear_property(&mut self, name: &str);
70
71    /// Adds a user-defined property to the message.
72    ///
73    /// # Errors
74    ///
75    /// Returns an error if the property name is reserved by the system or if the name or value is
76    /// empty.
77    fn put_user_property(&mut self, name: CheetahString, value: CheetahString) -> RocketMQResult<()> {
78        if name.is_empty() || value.is_empty() {
79            return Err(RocketMQError::InvalidProperty(
80                "The name or value of property can not be null or blank string!".to_string(),
81            ));
82        }
83        if STRING_HASH_SET.contains(name.as_str()) {
84            return Err(RocketMQError::InvalidProperty(format!(
85                "The Property<{name}> is used by system, input another please"
86            )));
87        }
88        self.put_property(name, value);
89        Ok(())
90    }
91
92    /// Retrieves a user-defined property value.
93    fn user_property(&self, name: &CheetahString) -> Option<CheetahString> {
94        self.property(name)
95    }
96
97    /// Retrieves a reference to a user-defined property value.
98    fn user_property_ref(&self, name: &CheetahString) -> Option<&CheetahString> {
99        self.property_ref(name)
100    }
101
102    /// Retrieves a property value.
103    fn property(&self, name: &CheetahString) -> Option<CheetahString>;
104
105    /// Retrieves a reference to a property value.
106    fn property_ref(&self, name: &CheetahString) -> Option<&CheetahString>;
107
108    /// Returns the topic of the message.
109    fn topic(&self) -> &CheetahString;
110
111    /// Sets the topic for the message.
112    fn set_topic(&mut self, topic: CheetahString);
113
114    /// Returns the tags associated with the message.
115    fn tags(&self) -> Option<CheetahString> {
116        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_TAGS))
117    }
118
119    /// Returns a reference to the tags associated with the message.
120    fn tags_ref(&self) -> Option<&CheetahString> {
121        self.property_ref(&CheetahString::from_static_str(MessageConst::PROPERTY_TAGS))
122    }
123
124    /// Sets the tags for the message.
125    fn set_tags(&mut self, tags: CheetahString) {
126        self.put_property(CheetahString::from_static_str(MessageConst::PROPERTY_TAGS), tags);
127    }
128
129    /// Returns the keys associated with the message.
130    fn get_keys(&self) -> Option<CheetahString> {
131        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_KEYS))
132    }
133    /// Returns a reference to the keys associated with the message.
134    fn get_keys_ref(&self) -> Option<&CheetahString> {
135        self.property_ref(&CheetahString::from_static_str(MessageConst::PROPERTY_KEYS))
136    }
137    /// Sets the message keys from a collection, joining them with spaces.
138    fn set_keys_from_collection(&mut self, key_collection: Vec<String>) {
139        let keys = key_collection.join(MessageConst::KEY_SEPARATOR);
140        self.set_keys(CheetahString::from_string(keys));
141    }
142
143    /// Returns the delay time level of the message, or 0 if not set.
144    fn delay_time_level(&self) -> i32 {
145        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_DELAY_TIME_LEVEL))
146            .and_then(|v| v.parse().ok())
147            .unwrap_or(0)
148    }
149
150    /// Sets the delay time level for the message.
151    fn set_delay_time_level(&mut self, level: i32) {
152        self.put_property(
153            CheetahString::from_static_str(MessageConst::PROPERTY_DELAY_TIME_LEVEL),
154            CheetahString::from_string(level.to_string()),
155        );
156    }
157
158    /// Returns whether the message should wait for store acknowledgment.
159    ///
160    /// Defaults to `true` if not set.
161    fn is_wait_store_msg_ok(&self) -> bool {
162        self.property(&CheetahString::from_static_str(
163            MessageConst::PROPERTY_WAIT_STORE_MSG_OK,
164        ))
165        .map(|v| v.as_str().eq_ignore_ascii_case("true"))
166        .unwrap_or(true)
167    }
168
169    /// Sets whether the message should wait for store acknowledgment.
170    fn set_wait_store_msg_ok(&mut self, wait_store_msg_ok: bool) {
171        self.put_property(
172            CheetahString::from_static_str(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
173            CheetahString::from_string(wait_store_msg_ok.to_string()),
174        );
175    }
176
177    /// Sets the instance ID for the message.
178    fn set_instance_id(&mut self, instance_id: CheetahString) {
179        self.put_property(
180            CheetahString::from_static_str(MessageConst::PROPERTY_INSTANCE_ID),
181            instance_id,
182        );
183    }
184
185    /// Returns the flag associated with the message.
186    fn get_flag(&self) -> i32;
187
188    /// Sets the flag for the message.
189    fn set_flag(&mut self, flag: i32);
190
191    /// Returns the body of the message.
192    fn get_body(&self) -> Option<&Bytes>;
193
194    /// Sets the body of the message.
195    fn set_body(&mut self, body: Bytes);
196
197    /// Returns all properties associated with the message.
198    fn get_properties(&self) -> &HashMap<CheetahString, CheetahString>;
199
200    /// Sets multiple properties for the message.
201    fn set_properties(&mut self, properties: HashMap<CheetahString, CheetahString>);
202
203    /// Returns the message priority, or `-1` if not set or not parseable.
204    fn priority(&self) -> i32 {
205        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_PRIORITY))
206            .and_then(|v| v.parse().ok())
207            .unwrap_or(-1)
208    }
209
210    /// Sets the message priority, matching Java's non-negative validation.
211    fn try_set_priority(&mut self, priority: i32) -> RocketMQResult<()> {
212        if priority < 0 {
213            return Err(RocketMQError::illegal_argument(
214                "The priority must be greater than or equal to 0",
215            ));
216        }
217        self.put_property(
218            CheetahString::from_static_str(MessageConst::PROPERTY_PRIORITY),
219            CheetahString::from_string(priority.to_string()),
220        );
221        Ok(())
222    }
223
224    /// Returns the buyer ID associated with the message.
225    fn buyer_id(&self) -> Option<CheetahString> {
226        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_BUYER_ID))
227    }
228
229    /// Returns a reference to the buyer ID associated with the message.
230    fn buyer_id_ref(&self) -> Option<&CheetahString> {
231        self.property_ref(&CheetahString::from_static_str(MessageConst::PROPERTY_BUYER_ID))
232    }
233
234    /// Sets the buyer ID for the message.
235    fn set_buyer_id(&mut self, buyer_id: CheetahString) {
236        self.put_property(
237            CheetahString::from_static_str(MessageConst::PROPERTY_BUYER_ID),
238            buyer_id,
239        );
240    }
241
242    /// Retrieves the transaction ID associated with the message.
243    ///
244    /// # Returns
245    ///
246    /// A reference to the transaction ID as a `&str`.
247    fn transaction_id(&self) -> Option<&CheetahString>;
248
249    /// Sets the transaction ID for the message.
250    fn set_transaction_id(&mut self, transaction_id: CheetahString);
251
252    /// Sets the delay time for the message in seconds.
253    fn set_delay_time_sec(&mut self, sec: u64) {
254        self.put_property(
255            CheetahString::from_static_str(MessageConst::PROPERTY_TIMER_DELAY_SEC),
256            CheetahString::from_string(sec.to_string()),
257        );
258    }
259
260    /// Returns the delay time for the message in seconds, or 0 if not set.
261    fn get_delay_time_sec(&self) -> u64 {
262        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_TIMER_DELAY_SEC))
263            .and_then(|v| v.parse().ok())
264            .unwrap_or(0)
265    }
266
267    /// Sets the delay time for the message in milliseconds.
268    fn set_delay_time_ms(&mut self, time_ms: u64) {
269        self.put_property(
270            CheetahString::from_static_str(MessageConst::PROPERTY_TIMER_DELAY_MS),
271            CheetahString::from_string(time_ms.to_string()),
272        );
273    }
274
275    /// Returns the delay time for the message in milliseconds, or 0 if not set.
276    fn get_delay_time_ms(&self) -> u64 {
277        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_TIMER_DELAY_MS))
278            .and_then(|v| v.parse().ok())
279            .unwrap_or(0)
280    }
281
282    /// Sets the delivery time for the message in milliseconds.
283    fn set_deliver_time_ms(&mut self, time_ms: u64) {
284        self.put_property(
285            CheetahString::from_static_str(MessageConst::PROPERTY_TIMER_DELIVER_MS),
286            CheetahString::from_string(time_ms.to_string()),
287        );
288    }
289
290    /// Returns the delivery time for the message in milliseconds, or 0 if not set.
291    fn get_deliver_time_ms(&self) -> u64 {
292        self.property(&CheetahString::from_static_str(MessageConst::PROPERTY_TIMER_DELIVER_MS))
293            .and_then(|v| v.parse().ok())
294            .unwrap_or(0)
295    }
296
297    /// Returns a mutable reference to the compressed body of the message.
298    fn get_compressed_body_mut(&mut self) -> Option<&mut Bytes>;
299
300    /// Returns a reference to the compressed body of the message.
301    fn get_compressed_body(&self) -> Option<&Bytes>;
302
303    /// Sets the compressed body of the message.
304    fn set_compressed_body_mut(&mut self, compressed_body: Bytes);
305
306    /// Takes ownership of the message body, leaving it empty.
307    fn take_body(&mut self) -> Option<Bytes>;
308
309    /// Returns a reference to the message as a trait object.
310    fn as_any(&self) -> &dyn Any;
311
312    /// Returns a mutable reference to the message as a trait object.
313    fn as_any_mut(&mut self) -> &mut dyn Any;
314}
315
316#[cfg(test)]
317mod tests {
318    use super::MessageConst;
319    use super::MessageTrait;
320    use super::STRING_HASH_SET;
321    use crate::common::message::message_builder::MessageBuilder;
322    use crate::common::message::message_single::Message;
323    use cheetah_string::CheetahString;
324
325    #[test]
326    fn trait_object_get_transaction_id_uses_default_compat_accessor() {
327        let mut message: Message = MessageBuilder::new()
328            .topic("TopicTest")
329            .body_slice(b"payload")
330            .build_unchecked();
331        MessageTrait::set_transaction_id(&mut message, CheetahString::from("tx-123"));
332
333        let message_trait: &dyn MessageTrait = &message;
334
335        assert_eq!(message_trait.transaction_id(), Some(&CheetahString::from("tx-123")));
336    }
337
338    #[test]
339    fn timer_engine_type_uses_java_protocol_key() {
340        assert_eq!(MessageConst::TIMER_ENGINE_TYPE, "E_T");
341        assert_eq!(MessageConst::TIMER_ENGINE_FILE_TIME_WHEEL, "F");
342        assert_eq!(MessageConst::TIMER_ENGINE_ROCKSDB_TIMELINE, "R");
343    }
344
345    #[test]
346    fn priority_constant_matches_java_and_is_reserved() {
347        assert_eq!(MessageConst::PROPERTY_PRIORITY, "_SYS_MSG_PRIORITY_");
348        assert!(STRING_HASH_SET.contains(MessageConst::PROPERTY_PRIORITY));
349    }
350
351    #[test]
352    fn message_const_missing_java_protocol_keys_are_aligned() {
353        assert_eq!(MessageConst::PROPERTY_ORIGIN_GROUP, "ORIGIN_GROUP");
354        assert_eq!(MessageConst::PROPERTY_TRANS_OFFSET, "TRANS_OFFSET");
355        assert_eq!(MessageConst::PROPERTY_TIMER_ROLL_LABEL, "TIMER_ROLL_LABEL");
356        assert!(STRING_HASH_SET.contains(MessageConst::PROPERTY_ORIGIN_GROUP));
357    }
358}
359
360/// Magic code for message format version 1.
361pub const MESSAGE_MAGIC_CODE_V1: i32 = -626843481;
362
363/// Magic code for message format version 2.
364pub const MESSAGE_MAGIC_CODE_V2: i32 = -626843477;
365
366/// Represents the message format version.
367#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Default)]
368pub enum MessageVersion {
369    #[default]
370    V1,
371    V2,
372}
373
374impl fmt::Display for MessageVersion {
375    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
376        match self {
377            MessageVersion::V1 => write!(f, "V1"),
378            MessageVersion::V2 => write!(f, "V2"),
379        }
380    }
381}
382
383impl MessageVersion {
384    /// Returns the message version corresponding to the given magic code.
385    ///
386    /// # Errors
387    ///
388    /// Returns an error if the magic code is not recognized.
389    pub fn value_of_magic_code(magic_code: i32) -> Result<MessageVersion, &'static str> {
390        match magic_code {
391            MESSAGE_MAGIC_CODE_V1 => Ok(MessageVersion::V1),
392            MESSAGE_MAGIC_CODE_V2 => Ok(MessageVersion::V2),
393            _ => Err("Invalid magicCode"),
394        }
395    }
396
397    /// Returns the magic code for this message version.
398    pub fn get_magic_code(&self) -> i32 {
399        match self {
400            MessageVersion::V1 => MESSAGE_MAGIC_CODE_V1,
401            MessageVersion::V2 => MESSAGE_MAGIC_CODE_V2,
402        }
403    }
404
405    /// Returns the number of bytes used to encode the topic length.
406    pub fn get_topic_length_size(&self) -> usize {
407        match self {
408            MessageVersion::V1 => 1,
409            MessageVersion::V2 => 2,
410        }
411    }
412
413    /// Reads and returns the topic length from the buffer, advancing the buffer position.
414    pub fn get_topic_length(&self, buffer: &mut Bytes) -> usize {
415        match self {
416            MessageVersion::V1 => buffer.get_u8() as usize,
417            MessageVersion::V2 => buffer.get_i16() as usize,
418        }
419    }
420
421    /// Returns the topic length from the buffer at the specified index without advancing the
422    /// position.
423    pub fn get_topic_length_at_index(&self, buffer: &[u8], index: usize) -> usize {
424        match self {
425            MessageVersion::V1 => buffer[index] as usize,
426            MessageVersion::V2 => ((buffer[index] as usize) << 8) | (buffer[index + 1] as usize),
427        }
428    }
429
430    /// Writes the topic length to the buffer according to the message version.
431    pub fn put_topic_length(&self, buffer: &mut Vec<u8>, topic_length: usize) {
432        match self {
433            MessageVersion::V1 => buffer.push(topic_length as u8),
434            MessageVersion::V2 => {
435                buffer.push((topic_length >> 8) as u8);
436                buffer.push((topic_length & 0xFF) as u8);
437            }
438        }
439    }
440
441    /// Returns `true` if this is message format version 1.
442    pub fn is_v1(&self) -> bool {
443        match self {
444            MessageVersion::V1 => true,
445            MessageVersion::V2 => false,
446        }
447    }
448
449    /// Returns `true` if this is message format version 2.
450    pub fn is_v2(&self) -> bool {
451        match self {
452            MessageVersion::V1 => false,
453            MessageVersion::V2 => true,
454        }
455    }
456}
457
458/// Defines constants for message property names and configuration values.
459pub struct MessageConst;
460
461impl MessageConst {
462    pub const DUP_INFO: &'static str = "DUP_INFO";
463    pub const KEY_SEPARATOR: &'static str = " ";
464    /// Host address where the message was born.
465    pub const PROPERTY_BORN_HOST: &'static str = "__BORNHOST";
466    /// Timestamp when the message was born.
467    pub const PROPERTY_BORN_TIMESTAMP: &'static str = "BORN_TIMESTAMP";
468    pub const PROPERTY_BUYER_ID: &'static str = "BUYER_ID";
469    /// Time in seconds during which transaction checks are suppressed.
470    pub const PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS: &'static str = "CHECK_IMMUNITY_TIME_IN_SECONDS";
471    pub const PROPERTY_CLUSTER: &'static str = "CLUSTER";
472    pub const PROPERTY_CONSUME_START_TIMESTAMP: &'static str = "CONSUME_START_TIME";
473    pub const PROPERTY_CORRECTION_FLAG: &'static str = "CORRECTION_FLAG";
474    pub const PROPERTY_CORRELATION_ID: &'static str = "CORRELATION_ID";
475    pub const PROPERTY_CRC32: &'static str = "__CRC32#";
476    pub const PROPERTY_DELAY_TIME_LEVEL: &'static str = "DELAY";
477    pub const PROPERTY_DLQ_ORIGIN_MESSAGE_ID: &'static str = "DLQ_ORIGIN_MESSAGE_ID";
478    pub const PROPERTY_STARTDE_LIVER_TIME: &'static str = "__STARTDELIVERTIME";
479    /// Original topic name for messages in the dead-letter queue.
480    pub const PROPERTY_DLQ_ORIGIN_TOPIC: &'static str = "DLQ_ORIGIN_TOPIC";
481    pub const PROPERTY_EXTEND_UNIQ_INFO: &'static str = "EXTEND_UNIQ_INFO";
482    pub const PROPERTY_FIRST_POP_TIME: &'static str = "1ST_POP_TIME";
483    pub const PROPERTY_FORWARD_QUEUE_ID: &'static str = "PROPERTY_FORWARD_QUEUE_ID";
484    pub const PROPERTY_INNER_BASE: &'static str = "INNER_BASE";
485    pub const PROPERTY_INNER_MULTI_DISPATCH: &'static str = "INNER_MULTI_DISPATCH";
486    pub const PROPERTY_INNER_MULTI_QUEUE_OFFSET: &'static str = "INNER_MULTI_QUEUE_OFFSET";
487    pub const PROPERTY_INNER_NUM: &'static str = "INNER_NUM";
488    pub const PROPERTY_INSTANCE_ID: &'static str = "INSTANCE_ID";
489    pub const PROPERTY_KEYS: &'static str = "KEYS";
490    pub const PROPERTY_MAX_OFFSET: &'static str = "MAX_OFFSET";
491    pub const PROPERTY_MAX_RECONSUME_TIMES: &'static str = "MAX_RECONSUME_TIMES";
492    pub const PROPERTY_MESSAGE_REPLY_TO_CLIENT: &'static str = "REPLY_TO_CLIENT";
493    pub const PROPERTY_MESSAGE_TTL: &'static str = "TTL";
494    pub const PROPERTY_MESSAGE_TYPE: &'static str = "MSG_TYPE";
495    pub const PROPERTY_MIN_OFFSET: &'static str = "MIN_OFFSET";
496    pub const PROPERTY_MQ2_FLAG: &'static str = "MQ2_FLAG";
497    pub const PROPERTY_MSG_REGION: &'static str = "MSG_REGION";
498    pub const PROPERTY_ORIGIN_GROUP: &'static str = "ORIGIN_GROUP";
499    pub const PROPERTY_ORIGIN_MESSAGE_ID: &'static str = "ORIGIN_MESSAGE_ID";
500    pub const PROPERTY_POP_CK: &'static str = "POP_CK";
501    pub const PROPERTY_POP_CK_OFFSET: &'static str = "POP_CK_OFFSET";
502    pub const PROPERTY_PRODUCER_GROUP: &'static str = "PGROUP";
503    pub const PROPERTY_PRIORITY: &'static str = "_SYS_MSG_PRIORITY_";
504    pub const PROPERTY_PUSH_REPLY_TIME: &'static str = "PUSH_REPLY_TIME";
505    pub const PROPERTY_REAL_QUEUE_ID: &'static str = "REAL_QID";
506    pub const PROPERTY_REAL_TOPIC: &'static str = "REAL_TOPIC";
507    pub const PROPERTY_RECONSUME_TIME: &'static str = "RECONSUME_TIME";
508    pub const PROPERTY_REDIRECT: &'static str = "REDIRECT";
509    pub const PROPERTY_REPLY_MESSAGE_ARRIVE_TIME: &'static str = "ARRIVE_TIME";
510    pub const PROPERTY_RETRY_TOPIC: &'static str = "RETRY_TOPIC";
511    pub const PROPERTY_SHARDING_KEY: &'static str = "__SHARDINGKEY";
512    pub const PROPERTY_LITE_TOPIC: &'static str = "__LITE_TOPIC";
513    pub const PROPERTY_TAGS: &'static str = "TAGS";
514    pub const PROPERTY_TIMER_DELAY_LEVEL: &'static str = "TIMER_DELAY_LEVEL";
515    pub const PROPERTY_TIMER_DELAY_MS: &'static str = "TIMER_DELAY_MS";
516    pub const PROPERTY_TIMER_DELAY_SEC: &'static str = "TIMER_DELAY_SEC";
517    pub const PROPERTY_TIMER_DELIVER_MS: &'static str = "TIMER_DELIVER_MS";
518    pub const PROPERTY_TIMER_DEL_UNIQKEY: &'static str = "TIMER_DEL_UNIQKEY";
519    pub const PROPERTY_TIMER_DEQUEUE_MS: &'static str = "TIMER_DEQUEUE_MS";
520    pub const PROPERTY_TIMER_ENQUEUE_MS: &'static str = "TIMER_ENQUEUE_MS";
521    pub const PROPERTY_TIMER_OUT_MS: &'static str = "TIMER_OUT_MS";
522    pub const PROPERTY_TIMER_ROLL_LABEL: &'static str = "TIMER_ROLL_LABEL";
523    pub const PROPERTY_TIMER_ROLL_TIMES: &'static str = "TIMER_ROLL_TIMES";
524    pub const PROPERTY_TRACE_CONTEXT: &'static str = "TRACE_CONTEXT";
525    pub const PROPERTY_TRACE_SWITCH: &'static str = "TRACE_ON";
526    pub const PROPERTY_TRANS_OFFSET: &'static str = "TRANS_OFFSET";
527    pub const PROPERTY_TRANSACTION_CHECK_TIMES: &'static str = "TRANSACTION_CHECK_TIMES";
528    pub const PROPERTY_TRANSACTION_ID: &'static str = "__transactionId__";
529    pub const PROPERTY_TRANSACTION_PREPARED: &'static str = "TRAN_MSG";
530    pub const PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET: &'static str = "TRAN_PREPARED_QUEUE_OFFSET";
531    pub const PROPERTY_TRANSFER_FLAG: &'static str = "TRANSFER_FLAG";
532    /// Transient property for group system flags set by the client when pulling messages.
533    pub const PROPERTY_TRANSIENT_GROUP_CONFIG: &'static str = "__RMQ.TRANSIENT.GROUP_SYS_FLAG";
534    /// Prefix for transient properties that are not persisted to broker disk.
535    pub const PROPERTY_TRANSIENT_PREFIX: &'static str = "__RMQ.TRANSIENT.";
536    /// Transient property for topic system flags set by the client when pulling messages.
537    pub const PROPERTY_TRANSIENT_TOPIC_CONFIG: &'static str = "__RMQ.TRANSIENT.TOPIC_SYS_FLAG";
538    pub const PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX: &'static str = "UNIQ_KEY";
539    pub const PROPERTY_WAIT_STORE_MSG_OK: &'static str = "WAIT";
540
541    /// Timer engine type identifier for RocksDB timeline implementation.
542    pub const TIMER_ENGINE_ROCKSDB_TIMELINE: &'static str = "R";
543    /// Timer engine type identifier for file-based time wheel implementation.
544    pub const TIMER_ENGINE_FILE_TIME_WHEEL: &'static str = "F";
545    /// Property name for timer engine type.
546    pub const TIMER_ENGINE_TYPE: &'static str = "E_T";
547
548    /// Index type identifier for message key indexing.
549    pub const INDEX_KEY_TYPE: &'static str = "K";
550    /// Index type identifier for unique indexing.
551    pub const INDEX_UNIQUE_TYPE: &'static str = "U";
552    /// Index type identifier for tag indexing.
553    pub const INDEX_TAG_TYPE: &'static str = "T";
554}
555
556/// Set of system-reserved property names that cannot be used as user-defined properties.
557pub static STRING_HASH_SET: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
558    let mut set = HashSet::with_capacity(64);
559    set.insert(MessageConst::PROPERTY_TRACE_SWITCH);
560    set.insert(MessageConst::PROPERTY_MSG_REGION);
561    set.insert(MessageConst::PROPERTY_KEYS);
562    set.insert(MessageConst::PROPERTY_TAGS);
563    set.insert(MessageConst::PROPERTY_WAIT_STORE_MSG_OK);
564    set.insert(MessageConst::PROPERTY_DELAY_TIME_LEVEL);
565    set.insert(MessageConst::PROPERTY_RETRY_TOPIC);
566    set.insert(MessageConst::PROPERTY_ORIGIN_GROUP);
567    set.insert(MessageConst::PROPERTY_REAL_TOPIC);
568    set.insert(MessageConst::PROPERTY_REAL_QUEUE_ID);
569    set.insert(MessageConst::PROPERTY_TRANSACTION_PREPARED);
570    set.insert(MessageConst::PROPERTY_PRODUCER_GROUP);
571    set.insert(MessageConst::PROPERTY_MIN_OFFSET);
572    set.insert(MessageConst::PROPERTY_MAX_OFFSET);
573    set.insert(MessageConst::PROPERTY_BUYER_ID);
574    set.insert(MessageConst::PROPERTY_ORIGIN_MESSAGE_ID);
575    set.insert(MessageConst::PROPERTY_TRANSFER_FLAG);
576    set.insert(MessageConst::PROPERTY_CORRECTION_FLAG);
577    set.insert(MessageConst::PROPERTY_MQ2_FLAG);
578    set.insert(MessageConst::PROPERTY_RECONSUME_TIME);
579    set.insert(MessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
580    set.insert(MessageConst::PROPERTY_MAX_RECONSUME_TIMES);
581    set.insert(MessageConst::PROPERTY_CONSUME_START_TIMESTAMP);
582    set.insert(MessageConst::PROPERTY_POP_CK);
583    set.insert(MessageConst::PROPERTY_POP_CK_OFFSET);
584    set.insert(MessageConst::PROPERTY_FIRST_POP_TIME);
585    set.insert(MessageConst::PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET);
586    set.insert(MessageConst::DUP_INFO);
587    set.insert(MessageConst::PROPERTY_EXTEND_UNIQ_INFO);
588    set.insert(MessageConst::PROPERTY_INSTANCE_ID);
589    set.insert(MessageConst::PROPERTY_CORRELATION_ID);
590    set.insert(MessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT);
591    set.insert(MessageConst::PROPERTY_MESSAGE_TTL);
592    set.insert(MessageConst::PROPERTY_REPLY_MESSAGE_ARRIVE_TIME);
593    set.insert(MessageConst::PROPERTY_PUSH_REPLY_TIME);
594    set.insert(MessageConst::PROPERTY_CLUSTER);
595    set.insert(MessageConst::PROPERTY_MESSAGE_TYPE);
596    set.insert(MessageConst::PROPERTY_INNER_MULTI_QUEUE_OFFSET);
597    set.insert(MessageConst::PROPERTY_TIMER_DELAY_MS);
598    set.insert(MessageConst::PROPERTY_TIMER_DELAY_SEC);
599    set.insert(MessageConst::PROPERTY_TIMER_DELIVER_MS);
600    set.insert(MessageConst::PROPERTY_TIMER_ENQUEUE_MS);
601    set.insert(MessageConst::PROPERTY_TIMER_DEQUEUE_MS);
602    set.insert(MessageConst::PROPERTY_TIMER_ROLL_TIMES);
603    set.insert(MessageConst::PROPERTY_TIMER_OUT_MS);
604    set.insert(MessageConst::PROPERTY_TIMER_DEL_UNIQKEY);
605    set.insert(MessageConst::PROPERTY_TIMER_DELAY_LEVEL);
606    set.insert(MessageConst::PROPERTY_BORN_HOST);
607    set.insert(MessageConst::PROPERTY_BORN_TIMESTAMP);
608    set.insert(MessageConst::PROPERTY_DLQ_ORIGIN_TOPIC);
609    set.insert(MessageConst::PROPERTY_DLQ_ORIGIN_MESSAGE_ID);
610    set.insert(MessageConst::PROPERTY_CRC32);
611    set.insert(MessageConst::PROPERTY_PRIORITY);
612    set.insert(MessageConst::PROPERTY_LITE_TOPIC);
613    set
614});
rocketmq_common/common/message.rs

rocketmq_common/common/
message.rs
