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

rocketmq_common/common/message/
message_property.rs
