rocketmq_model/common/message/
message_property.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::collections::HashMap;
16
17use cheetah_string::CheetahString;
18
19use crate::common::message::MessageConst;
20
21/// Type-safe message properties.
22#[derive(Clone, Debug, Default)]
23pub struct MessageProperties {
24    inner: HashMap<CheetahString, CheetahString>,
25}
26
27impl MessageProperties {
28    /// Creates a new empty properties collection.
29    pub fn new() -> Self {
30        Self::default()
31    }
32
33    /// Gets a property value by key.
34    #[inline]
35    pub fn get(&self, key: MessagePropertyKey) -> Option<&str> {
36        self.inner.get(key.as_str()).map(|s| s.as_str())
37    }
38
39    /// Sets a property value.
40    pub(crate) fn insert(&mut self, key: MessagePropertyKey, value: impl Into<CheetahString>) {
41        self.inner.insert(key.to_cheetah_string(), value.into());
42    }
43
44    /// Returns all properties as a map (for serialization and external crates).
45    #[inline]
46    pub fn as_map(&self) -> &HashMap<CheetahString, CheetahString> {
47        &self.inner
48    }
49
50    /// Returns all properties as a mutable map (for internal use).
51    #[doc(hidden)]
52    #[inline]
53    pub fn as_map_mut(&mut self) -> &mut HashMap<CheetahString, CheetahString> {
54        &mut self.inner
55    }
56
57    /// Creates MessageProperties from a HashMap.
58    #[inline]
59    pub fn from_map(map: HashMap<CheetahString, CheetahString>) -> Self {
60        Self { inner: map }
61    }
62
63    /// Returns the number of properties.
64    #[inline]
65    pub fn len(&self) -> usize {
66        self.inner.len()
67    }
68
69    /// Returns true if there are no properties.
70    #[inline]
71    pub fn is_empty(&self) -> bool {
72        self.inner.is_empty()
73    }
74
75    // Convenience accessors for well-known properties
76
77    /// Returns the tags, if any.
78    #[inline]
79    pub fn tags(&self) -> Option<&str> {
80        self.get(MessagePropertyKey::Tags)
81    }
82
83    /// Returns the keys as a vector, if any.
84    pub fn keys(&self) -> Option<Vec<String>> {
85        self.get(MessagePropertyKey::Keys).map(|s| {
86            s.split(MessageConst::KEY_SEPARATOR)
87                .filter(|k| !k.is_empty())
88                .map(|k| k.to_string())
89                .collect()
90        })
91    }
92
93    /// Returns the delay time level.
94    pub fn delay_level(&self) -> Option<i32> {
95        self.get(MessagePropertyKey::DelayTimeLevel)
96            .and_then(|s| s.parse().ok())
97    }
98
99    /// Returns the delay time in seconds.
100    pub fn delay_time_sec(&self) -> Option<u64> {
101        self.get(MessagePropertyKey::DelayTimeSec).and_then(|s| s.parse().ok())
102    }
103
104    /// Returns the delay time in milliseconds.
105    pub fn delay_time_ms(&self) -> Option<u64> {
106        self.get(MessagePropertyKey::DelayTimeMs).and_then(|s| s.parse().ok())
107    }
108
109    /// Returns the delivery time in milliseconds.
110    pub fn deliver_time_ms(&self) -> Option<u64> {
111        self.get(MessagePropertyKey::DeliverTimeMs).and_then(|s| s.parse().ok())
112    }
113
114    /// Returns the buyer ID.
115    pub fn buyer_id(&self) -> Option<&str> {
116        self.get(MessagePropertyKey::BuyerId)
117    }
118
119    /// Returns whether to wait for store confirmation.
120    pub fn wait_store_msg_ok(&self) -> bool {
121        self.get(MessagePropertyKey::WaitStoreMsgOk)
122            .map(|s| s.eq_ignore_ascii_case("true"))
123            .unwrap_or(true)
124    }
125
126    /// Returns the origin message ID.
127    pub fn origin_message_id(&self) -> Option<&str> {
128        self.get(MessagePropertyKey::OriginMessageId)
129    }
130
131    /// Returns the retry topic.
132    pub fn retry_topic(&self) -> Option<&str> {
133        self.get(MessagePropertyKey::RetryTopic)
134    }
135
136    /// Returns the real topic.
137    pub fn real_topic(&self) -> Option<&str> {
138        self.get(MessagePropertyKey::RealTopic)
139    }
140
141    /// Returns the real queue ID.
142    pub fn real_queue_id(&self) -> Option<i32> {
143        self.get(MessagePropertyKey::RealQueueId).and_then(|s| s.parse().ok())
144    }
145
146    /// Returns the unique client message ID.
147    pub fn unique_client_msg_id(&self) -> Option<&str> {
148        self.get(MessagePropertyKey::UniqueClientMsgId)
149    }
150
151    /// Returns the producer group.
152    pub fn producer_group(&self) -> Option<&str> {
153        self.get(MessagePropertyKey::ProducerGroup)
154    }
155
156    /// Returns the instance ID.
157    pub fn instance_id(&self) -> Option<&str> {
158        self.get(MessagePropertyKey::InstanceId)
159    }
160
161    /// Returns the correlation ID.
162    pub fn correlation_id(&self) -> Option<&str> {
163        self.get(MessagePropertyKey::CorrelationId)
164    }
165
166    /// Returns the message type.
167    pub fn message_type(&self) -> Option<&str> {
168        self.get(MessagePropertyKey::MessageType)
169    }
170
171    /// Returns the trace switch.
172    pub fn trace_switch(&self) -> bool {
173        self.get(MessagePropertyKey::TraceSwitch)
174            .map(|s| s == "true")
175            .unwrap_or(false)
176    }
177
178    /// Returns the message priority, or `-1` like Java when missing or invalid.
179    pub fn priority(&self) -> i32 {
180        self.get(MessagePropertyKey::Priority)
181            .and_then(|s| s.parse().ok())
182            .unwrap_or(-1)
183    }
184}
185
186/// Type-safe property keys.
187#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
188pub enum MessagePropertyKey {
189    Tags,
190    Keys,
191    WaitStoreMsgOk,
192    DelayTimeLevel,
193    DelayTimeSec,
194    DelayTimeMs,
195    DeliverTimeMs,
196    RetryTopic,
197    RealTopic,
198    RealQueueId,
199    TransactionPrepared,
200    ProducerGroup,
201    MinOffset,
202    MaxOffset,
203    BuyerId,
204    OriginGroup,
205    OriginMessageId,
206    TransferFlag,
207    CorrectionFlag,
208    MqReplyTopic,
209    MqReplyQueueId,
210    UniqueClientMsgId,
211    ReconsumeTime,
212    MsgRegion,
213    TraceSwitch,
214    UniqueKey,
215    MaxReconsumeTimes,
216    ConsumeStartTimestamp,
217    PopCk,
218    PopCkOffset,
219    FirstPopTime,
220    Priority,
221    TransactionPreparedQueueOffset,
222    DupInfo,
223    ExtendUniqInfo,
224    InstanceId,
225    CorrelationId,
226    MessageReplyToClient,
227    MessageTtl,
228    ReplyMessageArriveTime,
229    PushReplyTime,
230    Cluster,
231    MessageType,
232    InnerMultiQueueOffset,
233    TimerDelayLevel,
234    TimerEnqueueMs,
235    TimerDequeueMs,
236    TimerRollTimes,
237    TimerRollLabel,
238    TimerOutMs,
239    TimerDelUniqkey,
240    TimerOriginalDeliverMs,
241    TimerDeliveryToken,
242    TimerId,
243    TimerOwnerEpoch,
244    TimerLane,
245    TimerGeneration,
246    TimerEngineType,
247    TimerFormatVersion,
248    TimerPolicyFingerprint,
249    BornHost,
250    BornTimestamp,
251    DlqOriginTopic,
252    DlqOriginMessageId,
253    Crc32,
254    Redirect,
255    ForwardQueueId,
256    InnerBase,
257    InnerMultiDispatch,
258    InnerNum,
259    Mq2Flag,
260    TransOffset,
261    ShardingKey,
262    TransactionId,
263    TransactionCheckTimes,
264    TransientGroupConfig,
265    TransientTopicConfig,
266    TraceContext,
267}
268
269impl MessagePropertyKey {
270    /// Returns the string representation of this key.
271    pub fn as_str(&self) -> &'static str {
272        match self {
273            Self::Tags => MessageConst::PROPERTY_TAGS,
274            Self::Keys => MessageConst::PROPERTY_KEYS,
275            Self::WaitStoreMsgOk => MessageConst::PROPERTY_WAIT_STORE_MSG_OK,
276            Self::DelayTimeLevel => MessageConst::PROPERTY_DELAY_TIME_LEVEL,
277            Self::DelayTimeSec => MessageConst::PROPERTY_TIMER_DELAY_SEC,
278            Self::DelayTimeMs => MessageConst::PROPERTY_TIMER_DELAY_MS,
279            Self::DeliverTimeMs => MessageConst::PROPERTY_TIMER_DELIVER_MS,
280            Self::RetryTopic => MessageConst::PROPERTY_RETRY_TOPIC,
281            Self::RealTopic => MessageConst::PROPERTY_REAL_TOPIC,
282            Self::RealQueueId => MessageConst::PROPERTY_REAL_QUEUE_ID,
283            Self::TransactionPrepared => MessageConst::PROPERTY_TRANSACTION_PREPARED,
284            Self::ProducerGroup => MessageConst::PROPERTY_PRODUCER_GROUP,
285            Self::MinOffset => MessageConst::PROPERTY_MIN_OFFSET,
286            Self::MaxOffset => MessageConst::PROPERTY_MAX_OFFSET,
287            Self::BuyerId => MessageConst::PROPERTY_BUYER_ID,
288            Self::OriginGroup => MessageConst::PROPERTY_ORIGIN_GROUP,
289            Self::OriginMessageId => MessageConst::PROPERTY_ORIGIN_MESSAGE_ID,
290            Self::TransferFlag => MessageConst::PROPERTY_TRANSFER_FLAG,
291            Self::CorrectionFlag => MessageConst::PROPERTY_CORRECTION_FLAG,
292            Self::MqReplyTopic => MessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT,
293            Self::MqReplyQueueId => MessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT,
294            Self::UniqueClientMsgId => MessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
295            Self::ReconsumeTime => MessageConst::PROPERTY_RECONSUME_TIME,
296            Self::MsgRegion => MessageConst::PROPERTY_MSG_REGION,
297            Self::TraceSwitch => MessageConst::PROPERTY_TRACE_SWITCH,
298            Self::UniqueKey => MessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
299            Self::MaxReconsumeTimes => MessageConst::PROPERTY_MAX_RECONSUME_TIMES,
300            Self::ConsumeStartTimestamp => MessageConst::PROPERTY_CONSUME_START_TIMESTAMP,
301            Self::PopCk => MessageConst::PROPERTY_POP_CK,
302            Self::PopCkOffset => MessageConst::PROPERTY_POP_CK_OFFSET,
303            Self::FirstPopTime => MessageConst::PROPERTY_FIRST_POP_TIME,
304            Self::Priority => MessageConst::PROPERTY_PRIORITY,
305            Self::TransactionPreparedQueueOffset => MessageConst::PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET,
306            Self::DupInfo => MessageConst::DUP_INFO,
307            Self::ExtendUniqInfo => MessageConst::PROPERTY_EXTEND_UNIQ_INFO,
308            Self::InstanceId => MessageConst::PROPERTY_INSTANCE_ID,
309            Self::CorrelationId => MessageConst::PROPERTY_CORRELATION_ID,
310            Self::MessageReplyToClient => MessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT,
311            Self::MessageTtl => MessageConst::PROPERTY_MESSAGE_TTL,
312            Self::ReplyMessageArriveTime => MessageConst::PROPERTY_REPLY_MESSAGE_ARRIVE_TIME,
313            Self::PushReplyTime => MessageConst::PROPERTY_PUSH_REPLY_TIME,
314            Self::Cluster => MessageConst::PROPERTY_CLUSTER,
315            Self::MessageType => MessageConst::PROPERTY_MESSAGE_TYPE,
316            Self::InnerMultiQueueOffset => MessageConst::PROPERTY_INNER_MULTI_QUEUE_OFFSET,
317            Self::TimerDelayLevel => MessageConst::PROPERTY_TIMER_DELAY_LEVEL,
318            Self::TimerEnqueueMs => MessageConst::PROPERTY_TIMER_ENQUEUE_MS,
319            Self::TimerDequeueMs => MessageConst::PROPERTY_TIMER_DEQUEUE_MS,
320            Self::TimerRollTimes => MessageConst::PROPERTY_TIMER_ROLL_TIMES,
321            Self::TimerRollLabel => MessageConst::PROPERTY_TIMER_ROLL_LABEL,
322            Self::TimerOutMs => MessageConst::PROPERTY_TIMER_OUT_MS,
323            Self::TimerDelUniqkey => MessageConst::PROPERTY_TIMER_DEL_UNIQKEY,
324            Self::TimerOriginalDeliverMs => MessageConst::PROPERTY_TIMER_ORIGINAL_DELIVER_MS,
325            Self::TimerDeliveryToken => MessageConst::PROPERTY_TIMER_DELIVERY_TOKEN,
326            Self::TimerId => MessageConst::PROPERTY_TIMER_ID,
327            Self::TimerOwnerEpoch => MessageConst::PROPERTY_TIMER_OWNER_EPOCH,
328            Self::TimerLane => MessageConst::PROPERTY_TIMER_LANE,
329            Self::TimerGeneration => MessageConst::PROPERTY_TIMER_GENERATION,
330            Self::TimerEngineType => MessageConst::TIMER_ENGINE_TYPE,
331            Self::TimerFormatVersion => MessageConst::PROPERTY_TIMER_FORMAT_VERSION,
332            Self::TimerPolicyFingerprint => MessageConst::PROPERTY_TIMER_POLICY_FINGERPRINT,
333            Self::BornHost => MessageConst::PROPERTY_BORN_HOST,
334            Self::BornTimestamp => MessageConst::PROPERTY_BORN_TIMESTAMP,
335            Self::DlqOriginTopic => MessageConst::PROPERTY_DLQ_ORIGIN_TOPIC,
336            Self::DlqOriginMessageId => MessageConst::PROPERTY_DLQ_ORIGIN_MESSAGE_ID,
337            Self::Crc32 => MessageConst::PROPERTY_CRC32,
338            Self::Redirect => MessageConst::PROPERTY_REDIRECT,
339            Self::ForwardQueueId => MessageConst::PROPERTY_FORWARD_QUEUE_ID,
340            Self::InnerBase => MessageConst::PROPERTY_INNER_BASE,
341            Self::InnerMultiDispatch => MessageConst::PROPERTY_INNER_MULTI_DISPATCH,
342            Self::InnerNum => MessageConst::PROPERTY_INNER_NUM,
343            Self::Mq2Flag => MessageConst::PROPERTY_MQ2_FLAG,
344            Self::TransOffset => MessageConst::PROPERTY_TRANS_OFFSET,
345            Self::ShardingKey => MessageConst::PROPERTY_SHARDING_KEY,
346            Self::TransactionId => MessageConst::PROPERTY_TRANSACTION_ID,
347            Self::TransactionCheckTimes => MessageConst::PROPERTY_TRANSACTION_CHECK_TIMES,
348            Self::TransientGroupConfig => MessageConst::PROPERTY_TRANSIENT_GROUP_CONFIG,
349            Self::TransientTopicConfig => MessageConst::PROPERTY_TRANSIENT_TOPIC_CONFIG,
350            Self::TraceContext => MessageConst::PROPERTY_TRACE_CONTEXT,
351        }
352    }
353
354    pub(crate) fn to_cheetah_string(self) -> CheetahString {
355        CheetahString::from_static_str(self.as_str())
356    }
357}
358
359#[cfg(test)]
360mod tests {
361    use super::*;
362
363    #[test]
364    fn wait_store_msg_ok_matches_java_boolean_parse_semantics() {
365        let mut properties = MessageProperties::new();
366        assert!(properties.wait_store_msg_ok());
367
368        properties.insert(MessagePropertyKey::WaitStoreMsgOk, "true");
369        assert!(properties.wait_store_msg_ok());
370
371        properties.insert(MessagePropertyKey::WaitStoreMsgOk, "TRUE");
372        assert!(properties.wait_store_msg_ok());
373
374        properties.insert(MessagePropertyKey::WaitStoreMsgOk, "false");
375        assert!(!properties.wait_store_msg_ok());
376
377        properties.insert(MessagePropertyKey::WaitStoreMsgOk, "yes");
378        assert!(!properties.wait_store_msg_ok());
379    }
380
381    #[test]
382    fn priority_matches_java_default_and_parse_semantics() {
383        let mut properties = MessageProperties::new();
384        assert_eq!(properties.priority(), -1);
385
386        properties.insert(MessagePropertyKey::Priority, "7");
387        assert_eq!(properties.priority(), 7);
388
389        properties.insert(MessagePropertyKey::Priority, "invalid");
390        assert_eq!(properties.priority(), -1);
391    }
392
393    #[test]
394    fn protocol_property_keys_include_java_message_const_gaps() {
395        assert_eq!(MessagePropertyKey::OriginGroup.as_str(), "ORIGIN_GROUP");
396        assert_eq!(MessagePropertyKey::TransOffset.as_str(), "TRANS_OFFSET");
397        assert_eq!(MessagePropertyKey::TimerRollLabel.as_str(), "TIMER_ROLL_LABEL");
398    }
399}
rocketmq_model/common/message/message_property.rs

rocketmq_model/common/message/
message_property.rs
