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

rocketmq_model/common/
message.rs
