rocketmq_model/common/message/
message_accessor.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::message_single::Message;
20use crate::common::message::MessageConst;
21use crate::common::message::MessageTrait;
22
23pub struct MessageAccessor;
24
25impl MessageAccessor {
26    /// Sets the properties of a message.
27    ///
28    /// # Arguments
29    ///
30    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
31    /// * `properties` - A `HashMap` containing the properties to set.
32    #[inline]
33    pub fn set_properties<T: MessageTrait>(msg: &mut T, properties: HashMap<CheetahString, CheetahString>) {
34        msg.set_properties(properties);
35    }
36
37    /// Puts a property into a message.
38    ///
39    /// # Arguments
40    ///
41    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
42    /// * `name` - The name of the property.
43    /// * `value` - The value of the property.
44    #[inline]
45    pub fn put_property<T: MessageTrait>(msg: &mut T, name: CheetahString, value: CheetahString) {
46        msg.put_property(name, value);
47    }
48
49    /// Clears a property from a message.
50    ///
51    /// # Arguments
52    ///
53    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
54    /// * `name` - The name of the property to clear.
55    #[inline]
56    pub fn clear_property<T: MessageTrait>(msg: &mut T, name: &str) {
57        msg.clear_property(name);
58    }
59
60    /// Sets the transfer flag of a message.
61    ///
62    /// # Arguments
63    ///
64    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
65    /// * `unit` - The transfer flag value.
66    #[inline]
67    pub fn set_transfer_flag<T: MessageTrait>(msg: &mut T, unit: CheetahString) {
68        msg.put_property(
69            CheetahString::from_static_str(MessageConst::PROPERTY_TRANSFER_FLAG),
70            unit,
71        );
72    }
73
74    /// Gets the transfer flag of a message.
75    ///
76    /// # Arguments
77    ///
78    /// * `msg` - A reference to a message implementing the `MessageTrait`.
79    ///
80    /// # Returns
81    ///
82    /// * `Option<String>` - The transfer flag value if it exists.
83    #[inline]
84    pub fn get_transfer_flag<T: MessageTrait>(msg: &T) -> Option<CheetahString> {
85        msg.property(&CheetahString::from_static_str(MessageConst::PROPERTY_TRANSFER_FLAG))
86    }
87
88    /// Sets the correction flag of a message.
89    ///
90    /// # Arguments
91    ///
92    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
93    /// * `unit` - The correction flag value.
94    #[inline]
95    pub fn set_correction_flag<T: MessageTrait>(msg: &mut T, unit: CheetahString) {
96        msg.put_property(
97            CheetahString::from_static_str(MessageConst::PROPERTY_CORRECTION_FLAG),
98            unit,
99        );
100    }
101
102    /// Gets the correction flag of a message.
103    ///
104    /// # Arguments
105    ///
106    /// * `msg` - A reference to a message implementing the `MessageTrait`.
107    ///
108    /// # Returns
109    ///
110    /// * `Option<String>` - The correction flag value if it exists.
111    #[inline]
112    pub fn get_correction_flag<T: MessageTrait>(msg: &T) -> Option<CheetahString> {
113        msg.property(&CheetahString::from_static_str(MessageConst::PROPERTY_CORRECTION_FLAG))
114    }
115
116    /// Sets the origin message ID of a message.
117    ///
118    /// # Arguments
119    ///
120    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
121    /// * `origin_message_id` - The origin message ID value.
122    #[inline]
123    pub fn set_origin_message_id<T: MessageTrait>(msg: &mut T, origin_message_id: CheetahString) {
124        msg.put_property(
125            CheetahString::from_static_str(MessageConst::PROPERTY_ORIGIN_MESSAGE_ID),
126            origin_message_id,
127        );
128    }
129
130    /// Gets the origin message ID of a message.
131    ///
132    /// # Arguments
133    ///
134    /// * `msg` - A reference to a message implementing the `MessageTrait`.
135    ///
136    /// # Returns
137    ///
138    /// * `Option<String>` - The origin message ID value if it exists.
139    #[inline]
140    pub fn get_origin_message_id<T: MessageTrait>(msg: &T) -> Option<CheetahString> {
141        msg.property(&CheetahString::from_static_str(
142            MessageConst::PROPERTY_ORIGIN_MESSAGE_ID,
143        ))
144    }
145
146    /// Sets the MQ2 flag of a message.
147    ///
148    /// # Arguments
149    ///
150    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
151    /// * `flag` - The MQ2 flag value.
152    #[inline]
153    pub fn set_mq2_flag<T: MessageTrait>(msg: &mut T, flag: CheetahString) {
154        msg.put_property(CheetahString::from_static_str(MessageConst::PROPERTY_MQ2_FLAG), flag);
155    }
156
157    /// Gets the MQ2 flag of a message.
158    ///
159    /// # Arguments
160    ///
161    /// * `msg` - A reference to a message implementing the `MessageTrait`.
162    ///
163    /// # Returns
164    ///
165    /// * `Option<String>` - The MQ2 flag value if it exists.
166    #[inline]
167    pub fn get_mq2_flag<T: MessageTrait>(msg: &T) -> Option<CheetahString> {
168        msg.property(&CheetahString::from_static_str(MessageConst::PROPERTY_MQ2_FLAG))
169    }
170
171    /// Sets the reconsume time of a message.
172    ///
173    /// # Arguments
174    ///
175    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
176    /// * `reconsume_times` - The reconsume time value.
177    #[inline]
178    pub fn set_reconsume_time<T: MessageTrait>(msg: &mut T, reconsume_times: CheetahString) {
179        msg.put_property(
180            CheetahString::from_static_str(MessageConst::PROPERTY_RECONSUME_TIME),
181            reconsume_times,
182        );
183    }
184
185    /// Gets the reconsume time of a message.
186    ///
187    /// # Arguments
188    ///
189    /// * `msg` - A reference to a message implementing the `MessageTrait`.
190    ///
191    /// # Returns
192    ///
193    /// * `Option<String>` - The reconsume time value if it exists.
194    #[inline]
195    pub fn get_reconsume_time<T: MessageTrait>(msg: &T) -> Option<CheetahString> {
196        msg.property(&CheetahString::from_static_str(MessageConst::PROPERTY_RECONSUME_TIME))
197    }
198
199    /// Sets the maximum reconsume times of a message.
200    ///
201    /// # Arguments
202    ///
203    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
204    /// * `max_reconsume_times` - The maximum reconsume times value.
205    #[inline]
206    pub fn set_max_reconsume_times<T: MessageTrait>(msg: &mut T, max_reconsume_times: CheetahString) {
207        msg.put_property(
208            CheetahString::from_static_str(MessageConst::PROPERTY_MAX_RECONSUME_TIMES),
209            max_reconsume_times,
210        );
211    }
212
213    /// Gets the maximum reconsume times of a message.
214    ///
215    /// # Arguments
216    ///
217    /// * `msg` - A reference to a message implementing the `MessageTrait`.
218    ///
219    /// # Returns
220    ///
221    /// * `Option<String>` - The maximum reconsume times value if it exists.
222    #[inline]
223    pub fn get_max_reconsume_times<T: MessageTrait>(msg: &T) -> Option<CheetahString> {
224        msg.property(&CheetahString::from_static_str(
225            MessageConst::PROPERTY_MAX_RECONSUME_TIMES,
226        ))
227    }
228
229    /// Sets the consume start timestamp of a message.
230    ///
231    /// # Arguments
232    ///
233    /// * `msg` - A mutable reference to a message implementing the `MessageTrait`.
234    /// * `property_consume_start_time_stamp` - The consume start timestamp value.
235    #[inline]
236    pub fn set_consume_start_time_stamp<T: MessageTrait>(
237        msg: &mut T,
238        property_consume_start_time_stamp: CheetahString,
239    ) {
240        msg.put_property(
241            CheetahString::from_static_str(MessageConst::PROPERTY_CONSUME_START_TIMESTAMP),
242            property_consume_start_time_stamp,
243        );
244    }
245
246    /// Gets the consume start timestamp of a message.
247    ///
248    /// # Arguments
249    ///
250    /// * `msg` - A reference to a message implementing the `MessageTrait`.
251    ///
252    /// # Returns
253    ///
254    /// * `Option<String>` - The consume start timestamp value if it exists.
255    #[inline]
256    pub fn get_consume_start_time_stamp<T: MessageTrait>(msg: &T) -> Option<CheetahString> {
257        msg.property(&CheetahString::from_static_str(
258            MessageConst::PROPERTY_CONSUME_START_TIMESTAMP,
259        ))
260    }
261
262    /// Clones a message.
263    pub fn clone_message<M>(message: &M) -> Message
264    where
265        M: MessageTrait,
266    {
267        let mut new_message = Message::default();
268        new_message.set_topic(message.topic().clone());
269        if let Some(body) = message.get_body() {
270            new_message.set_body(Some(body.clone()));
271        }
272        new_message.set_flag(message.get_flag());
273        new_message.set_properties(message.get_properties().clone());
274        new_message
275    }
276}
277
278#[cfg(test)]
279mod tests {
280    use super::*;
281    use bytes::Bytes;
282
283    #[test]
284    fn clone_message_preserves_fields_and_copies_properties() {
285        let mut source = Message::default();
286        source.set_topic("topic".into());
287        source.set_body(Some(Bytes::from_static(b"body")));
288        source.set_flag(7);
289        source.put_property("key".into(), "value".into());
290        source.put_property("other".into(), "retained".into());
291
292        let cloned = MessageAccessor::clone_message(&source);
293        assert_eq!(cloned.topic(), source.topic());
294        assert_eq!(cloned.get_body(), source.get_body());
295        assert_eq!(cloned.get_flag(), source.get_flag());
296        assert_eq!(cloned.get_properties(), source.get_properties());
297
298        source.put_property("key".into(), "changed".into());
299        assert_eq!(cloned.property("key"), Some("value"));
300    }
301
302    #[test]
303    fn clone_message_without_body_retains_topic_and_properties() {
304        let mut source = Message::default();
305        source.set_topic("topic".into());
306        source.put_property("key".into(), "value".into());
307
308        let cloned = MessageAccessor::clone_message(&source);
309        assert_eq!(cloned.get_body(), None);
310        assert_eq!(cloned.topic(), source.topic());
311        assert_eq!(cloned.get_properties(), source.get_properties());
312    }
313}
rocketmq_model/common/message/message_accessor.rs

rocketmq_model/common/message/
message_accessor.rs
