1use std::any::Any;
16use std::collections::HashMap;
17use std::fmt;
18use std::fmt::Display;
19use std::fmt::Formatter;
20
21use bytes::Bytes;
22use cheetah_string::CheetahString;
23
24use crate::common::hasher::string_hasher::JavaStringHasher;
25use crate::common::message::message_body::MessageBody;
26use crate::common::message::message_builder::MessageBuilder;
27use crate::common::message::message_flag::MessageFlag;
28use crate::common::message::message_property::MessageProperties;
29use crate::common::message::MessageConst;
30use crate::common::message::MessageTrait;
31use crate::common::sys_flag::message_sys_flag::MessageSysFlag;
32use crate::common::TopicFilterType;
33
34#[derive(Clone)]
35pub struct Message {
36 topic: CheetahString,
37 flag: MessageFlag,
38 properties: MessageProperties,
39 body: MessageBody,
40 transaction_id: Option<CheetahString>,
41}
42
43impl Message {
45 pub(crate) fn from_builder(
46 topic: CheetahString,
47 body: MessageBody,
48 properties: MessageProperties,
49 flag: MessageFlag,
50 transaction_id: Option<CheetahString>,
51 ) -> Self {
52 Self {
53 topic,
54 flag,
55 properties,
56 body,
57 transaction_id,
58 }
59 }
60}
61
62impl Default for Message {
63 fn default() -> Self {
64 Self {
65 topic: CheetahString::new(),
66 flag: MessageFlag::empty(),
67 properties: MessageProperties::new(),
68 body: MessageBody::empty(),
69 transaction_id: None,
70 }
71 }
72}
73
74impl Message {
75 pub fn builder() -> MessageBuilder {
91 MessageBuilder::new()
92 }
93
94 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
96 #[allow(deprecated)]
97 #[inline]
98 pub fn new(topic: impl Into<CheetahString>, body: &[u8]) -> Self {
99 Self::with_details(topic, CheetahString::empty(), CheetahString::empty(), 0, body, true)
100 }
101
102 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
104 #[allow(deprecated)]
105 #[inline]
106 pub fn new_with_bytes(topic: impl Into<CheetahString>, body: Bytes) -> Self {
107 Self::with_details_bytes(topic, CheetahString::empty(), CheetahString::empty(), 0, body, true)
108 }
109
110 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
112 #[allow(deprecated)]
113 #[inline]
114 pub fn new_with_vec(topic: impl Into<CheetahString>, body: Vec<u8>) -> Self {
115 Self::with_details_bytes(
116 topic,
117 CheetahString::empty(),
118 CheetahString::empty(),
119 0,
120 Bytes::from(body),
121 true,
122 )
123 }
124
125 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
126 #[allow(deprecated)]
127 #[inline]
128 pub fn new_body(topic: impl Into<CheetahString>, body: Option<Bytes>) -> Self {
129 Self::with_details_body(topic, CheetahString::empty(), CheetahString::empty(), 0, body, true)
130 }
131
132 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
134 #[allow(deprecated)]
135 #[inline]
136 pub fn with_tags(topic: impl Into<CheetahString>, tags: impl Into<CheetahString>, body: &[u8]) -> Self {
137 Self::with_details(topic, tags, CheetahString::empty(), 0, body, true)
138 }
139
140 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
142 #[allow(deprecated)]
143 #[inline]
144 pub fn with_tags_bytes(topic: impl Into<CheetahString>, tags: impl Into<CheetahString>, body: Bytes) -> Self {
145 Self::with_details_bytes(topic, tags, CheetahString::empty(), 0, body, true)
146 }
147
148 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
150 #[allow(deprecated)]
151 #[inline]
152 pub fn with_keys(
153 topic: impl Into<CheetahString>,
154 tags: impl Into<CheetahString>,
155 keys: impl Into<CheetahString>,
156 body: &[u8],
157 ) -> Self {
158 Self::with_details(topic, tags, keys, 0, body, true)
159 }
160
161 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
163 #[allow(deprecated)]
164 #[inline]
165 pub fn with_keys_bytes(
166 topic: impl Into<CheetahString>,
167 tags: impl Into<CheetahString>,
168 keys: impl Into<CheetahString>,
169 body: Bytes,
170 ) -> Self {
171 Self::with_details_bytes(topic, tags, keys, 0, body, true)
172 }
173
174 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
176 #[allow(deprecated)]
177 pub fn with_details(
178 topic: impl Into<CheetahString>,
179 tags: impl Into<CheetahString>,
180 keys: impl Into<CheetahString>,
181 flag: i32,
182 body: &[u8],
183 wait_store_msg_ok: bool,
184 ) -> Self {
185 Self::with_details_bytes(topic, tags, keys, flag, Bytes::copy_from_slice(body), wait_store_msg_ok)
186 }
187
188 #[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
190 pub fn with_details_bytes(
191 topic: impl Into<CheetahString>,
192 tags: impl Into<CheetahString>,
193 keys: impl Into<CheetahString>,
194 flag: i32,
195 body: Bytes,
196 wait_store_msg_ok: bool,
197 ) -> Self {
198 let topic = topic.into();
199 let tags = tags.into();
200 let keys = keys.into();
201
202 let has_tags = !tags.is_empty();
204 let has_keys = !keys.is_empty();
205 let initial_capacity = (has_tags as usize) + (has_keys as usize) + 1;
206 let mut properties = HashMap::with_capacity(initial_capacity);
207
208 if has_tags {
210 properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_TAGS), tags);
211 }
212
213 if has_keys {
214 properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_KEYS), keys);
215 }
216
217 properties.insert(
218 CheetahString::from_static_str(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
219 CheetahString::from_static_str(if wait_store_msg_ok { "true" } else { "false" }),
220 );
221
222 Message {
223 topic,
224 flag: MessageFlag::from_bits(flag),
225 properties: MessageProperties::from_map(properties),
226 body: MessageBody::from(body),
227 transaction_id: None,
228 }
229 }
230
231 pub fn with_details_body(
232 topic: impl Into<CheetahString>,
233 tags: impl Into<CheetahString>,
234 keys: impl Into<CheetahString>,
235 flag: i32,
236 body: Option<Bytes>,
237 wait_store_msg_ok: bool,
238 ) -> Self {
239 let topic = topic.into();
240 let tags = tags.into();
241 let keys = keys.into();
242
243 let has_tags = !tags.is_empty();
245 let has_keys = !keys.is_empty();
246 let initial_capacity = (has_tags as usize) + (has_keys as usize) + 1;
247 let mut properties = HashMap::with_capacity(initial_capacity);
248
249 if has_tags {
250 properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_TAGS), tags);
251 }
252
253 if has_keys {
254 properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_KEYS), keys);
255 }
256
257 properties.insert(
258 CheetahString::from_static_str(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
259 CheetahString::from_static_str(if wait_store_msg_ok { "true" } else { "false" }),
260 );
261
262 Message {
263 topic,
264 flag: MessageFlag::from_bits(flag),
265 properties: MessageProperties::from_map(properties),
266 body: if let Some(b) = body {
267 MessageBody::from(b)
268 } else {
269 MessageBody::empty()
270 },
271 transaction_id: None,
272 }
273 }
274
275 #[inline]
276 pub fn set_tags(&mut self, tags: CheetahString) {
277 self.properties
278 .as_map_mut()
279 .insert(CheetahString::from_static_str(MessageConst::PROPERTY_TAGS), tags);
280 }
281
282 #[inline]
283 pub fn set_keys(&mut self, keys: CheetahString) {
284 self.properties
285 .as_map_mut()
286 .insert(CheetahString::from_static_str(MessageConst::PROPERTY_KEYS), keys);
287 }
288
289 #[inline]
290 pub fn clear_property(&mut self, name: impl Into<CheetahString>) {
291 self.properties.as_map_mut().remove(name.into().as_str());
292 }
293
294 #[inline]
295 pub fn set_properties(&mut self, properties: HashMap<CheetahString, CheetahString>) {
296 self.properties = MessageProperties::from_map(properties);
297 }
298
299 #[inline]
300 pub fn get_property(&self, key: &CheetahString) -> Option<CheetahString> {
301 self.properties.as_map().get(key).cloned()
302 }
303
304 #[inline]
305 pub fn body(&self) -> Option<bytes::Bytes> {
306 self.body.raw().cloned()
307 }
308
309 #[inline]
310 pub fn flag(&self) -> i32 {
311 self.flag.bits()
312 }
313
314 #[inline]
315 pub fn topic(&self) -> &CheetahString {
316 &self.topic
317 }
318
319 #[inline]
320 pub fn properties(&self) -> &MessageProperties {
321 &self.properties
322 }
323
324 #[inline]
325 pub fn transaction_id(&self) -> Option<&str> {
326 self.transaction_id.as_deref()
327 }
328
329 #[inline]
330 pub fn get_transaction_id(&self) -> Option<&CheetahString> {
331 self.transaction_id.as_ref()
332 }
333
334 #[inline]
335 pub fn get_tags(&self) -> Option<CheetahString> {
336 self.properties.as_map().get(MessageConst::PROPERTY_TAGS).cloned()
337 }
338
339 #[inline]
340 pub fn is_wait_store_msg_ok(&self) -> bool {
341 self.properties.wait_store_msg_ok()
342 }
343
344 #[inline]
345 pub fn delay_time_level(&self) -> i32 {
346 self.properties.delay_level().unwrap_or(0)
347 }
348
349 #[inline]
350 pub fn set_delay_time_level(&mut self, level: i32) {
351 self.properties.as_map_mut().insert(
352 CheetahString::from_static_str(MessageConst::PROPERTY_DELAY_TIME_LEVEL),
353 CheetahString::from(level.to_string()),
354 );
355 }
356
357 #[inline]
358 pub fn get_user_property(&self, name: impl Into<CheetahString>) -> Option<CheetahString> {
359 self.properties.as_map().get(name.into().as_str()).cloned()
360 }
361
362 #[inline]
363 pub fn as_any(&self) -> &dyn Any {
364 self
365 }
366
367 #[inline]
368 pub fn set_instance_id(&mut self, instance_id: impl Into<CheetahString>) {
369 self.properties.as_map_mut().insert(
370 CheetahString::from_static_str(MessageConst::PROPERTY_INSTANCE_ID),
371 instance_id.into(),
372 );
373 }
374
375 #[inline]
381 pub fn body_slice(&self) -> &[u8] {
382 self.body.as_slice()
383 }
384
385 #[inline]
387 pub fn into_body(self) -> MessageBody {
388 self.body
389 }
390
391 #[inline]
393 pub fn tags(&self) -> Option<&str> {
394 self.properties.tags()
395 }
396
397 #[inline]
399 pub fn keys(&self) -> Option<Vec<String>> {
400 self.properties.keys()
401 }
402
403 #[inline]
405 pub fn property(&self, key: &str) -> Option<&str> {
406 self.properties.as_map().get(key).map(|s| s.as_str())
407 }
408
409 #[inline]
411 pub fn message_flag(&self) -> MessageFlag {
412 self.flag
413 }
414
415 #[inline]
417 pub fn wait_store_msg_ok(&self) -> bool {
418 self.is_wait_store_msg_ok()
419 }
420
421 #[inline]
423 pub fn delay_level(&self) -> i32 {
424 self.delay_time_level()
425 }
426
427 #[inline]
429 pub fn buyer_id(&self) -> Option<&str> {
430 self.properties.buyer_id()
431 }
432
433 #[inline]
435 pub fn priority(&self) -> i32 {
436 self.properties.priority()
437 }
438
439 #[inline]
441 pub fn try_set_priority(&mut self, priority: i32) -> rocketmq_error::Result<()> {
442 if priority < 0 {
443 return Err(crate::error::invalid_argument(
444 "The priority must be greater than or equal to 0",
445 ));
446 }
447 self.properties.as_map_mut().insert(
448 CheetahString::from_static_str(MessageConst::PROPERTY_PRIORITY),
449 CheetahString::from(priority.to_string()),
450 );
451 Ok(())
452 }
453
454 #[inline]
456 pub fn instance_id(&self) -> Option<&str> {
457 self.properties.instance_id()
458 }
459
460 #[doc(hidden)]
464 #[inline]
465 pub fn topic_mut(&mut self) -> &mut CheetahString {
466 &mut self.topic
467 }
468
469 #[doc(hidden)]
471 #[inline]
472 pub fn flag_mut(&mut self) -> &mut MessageFlag {
473 &mut self.flag
474 }
475
476 #[doc(hidden)]
478 #[inline]
479 pub fn properties_mut(&mut self) -> &mut MessageProperties {
480 &mut self.properties
481 }
482
483 #[doc(hidden)]
485 #[inline]
486 pub fn body_mut(&mut self) -> &mut MessageBody {
487 &mut self.body
488 }
489
490 #[doc(hidden)]
492 #[inline]
493 pub fn transaction_id_mut(&mut self) -> &mut Option<CheetahString> {
494 &mut self.transaction_id
495 }
496
497 #[doc(hidden)]
499 #[inline]
500 pub fn set_topic(&mut self, topic: CheetahString) {
501 self.topic = topic;
502 }
503
504 #[doc(hidden)]
506 #[inline]
507 pub fn set_flag(&mut self, flag: i32) {
508 self.flag = MessageFlag::from_bits(flag);
509 }
510
511 #[doc(hidden)]
513 #[inline]
514 pub fn set_body(&mut self, body: Option<Bytes>) {
515 self.body = if let Some(b) = body {
516 MessageBody::from(b)
517 } else {
518 MessageBody::empty()
519 };
520 }
521
522 #[doc(hidden)]
524 #[inline]
525 pub fn take_body(&mut self) -> Option<Bytes> {
526 let old = std::mem::take(&mut self.body);
527 old.raw().cloned()
528 }
529
530 #[doc(hidden)]
532 #[inline]
533 pub fn compressed_body(&self) -> Option<&Bytes> {
534 self.body.compressed()
535 }
536}
537
538impl fmt::Debug for Message {
539 fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
540 f.debug_struct("Message")
541 .field("topic", &self.topic)
542 .field("flag", &self.flag.bits())
543 .field("body_len", &self.body_slice().len())
544 .field("property_count", &self.properties.len())
545 .field("transaction_id_present", &self.transaction_id.is_some())
546 .finish()
547 }
548}
549
550impl Display for Message {
551 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
552 write!(
553 f,
554 "Message{{topic='{}', flag={}, bodyLen={}, propertyCount={}, transactionIdPresent={}}}",
555 self.topic,
556 self.flag.bits(),
557 self.body_slice().len(),
558 self.properties.len(),
559 self.transaction_id.is_some()
560 )
561 }
562}
563
564#[allow(unused_variables)]
565impl MessageTrait for Message {
566 #[inline]
567 fn into_message(self) -> Message {
568 self
569 }
570
571 #[inline]
572 fn put_property(&mut self, key: CheetahString, value: CheetahString) {
573 self.properties.as_map_mut().insert(key, value);
574 }
575
576 #[inline]
577 fn clear_property(&mut self, name: &str) {
578 self.properties.as_map_mut().remove(name);
579 }
580
581 #[inline]
582 fn property(&self, name: &CheetahString) -> Option<CheetahString> {
583 self.properties.as_map().get(name).cloned()
584 }
585
586 fn property_ref(&self, name: &CheetahString) -> Option<&CheetahString> {
587 self.properties.as_map().get(name)
588 }
589
590 #[inline]
591 fn topic(&self) -> &CheetahString {
592 &self.topic
593 }
594
595 #[inline]
596 fn set_topic(&mut self, topic: CheetahString) {
597 self.set_topic(topic);
598 }
599
600 #[inline]
601 fn get_flag(&self) -> i32 {
602 self.flag.bits()
603 }
604
605 #[inline]
606 fn set_flag(&mut self, flag: i32) {
607 self.set_flag(flag);
608 }
609
610 #[inline]
611 fn get_body(&self) -> Option<&Bytes> {
612 self.body.raw()
613 }
614
615 #[inline]
616 fn set_body(&mut self, body: Bytes) {
617 self.set_body(Some(body));
618 }
619
620 #[inline]
621 fn get_properties(&self) -> &HashMap<CheetahString, CheetahString> {
622 self.properties.as_map()
623 }
624
625 #[inline]
626 fn set_properties(&mut self, properties: HashMap<CheetahString, CheetahString>) {
627 self.properties = MessageProperties::from_map(properties);
628 }
629
630 #[inline]
631 fn transaction_id(&self) -> Option<&CheetahString> {
632 self.transaction_id.as_ref()
633 }
634
635 #[inline]
636 fn set_transaction_id(&mut self, transaction_id: CheetahString) {
637 *self.transaction_id_mut() = Some(transaction_id);
638 }
639
640 #[inline]
641 fn get_compressed_body_mut(&mut self) -> Option<&mut Bytes> {
642 self.body.compressed_mut().as_mut()
643 }
644
645 #[inline]
646 fn get_compressed_body(&self) -> Option<&Bytes> {
647 self.compressed_body()
648 }
649
650 #[inline]
651 fn set_compressed_body_mut(&mut self, compressed_body: Bytes) {
652 self.body.set_compressed(compressed_body);
653 }
654
655 #[inline]
656 fn take_body(&mut self) -> Option<Bytes> {
657 self.take_body()
658 }
659
660 #[inline]
661 fn as_any(&self) -> &dyn Any {
662 self
663 }
664
665 #[inline]
666 fn as_any_mut(&mut self) -> &mut dyn Any {
667 self
668 }
669}
670
671pub fn parse_topic_filter_type(sys_flag: i32) -> TopicFilterType {
672 if (sys_flag & MessageSysFlag::MULTI_TAGS_FLAG) == MessageSysFlag::MULTI_TAGS_FLAG {
673 TopicFilterType::MultiTag
674 } else {
675 TopicFilterType::SingleTag
676 }
677}
678
679pub fn tags_string2tags_code(tags: Option<&CheetahString>) -> i64 {
680 match tags {
681 Some(tags) if !tags.is_empty() => JavaStringHasher::hash_str(tags.as_str()) as i64,
682 _ => 0,
683 }
684}
685
686#[cfg(test)]
687#[allow(deprecated)]
688mod tests {
689 use bytes::Bytes;
690
691 use super::*;
692
693 #[test]
694 fn test_message_new() {
695 let msg = Message::new("test_topic", b"test_body");
696 assert_eq!(msg.topic().as_str(), "test_topic");
697 assert_eq!(msg.body().unwrap().as_ref(), b"test_body");
698 }
699
700 #[test]
701 fn test_message_new_with_bytes() {
702 let body = Bytes::from_static(b"test_body");
703 let msg = Message::new_with_bytes("test_topic", body.clone());
704 assert_eq!(msg.topic().as_str(), "test_topic");
705 assert_eq!(msg.body().unwrap(), body);
706 }
707
708 #[test]
709 fn test_message_new_with_vec() {
710 let body = vec![1u8, 2, 3, 4, 5];
711 let msg = Message::new_with_vec("test_topic", body.clone());
712 assert_eq!(msg.topic().as_str(), "test_topic");
713 assert_eq!(msg.body().unwrap().as_ref(), body.as_slice());
714 }
715
716 #[test]
717 fn test_message_with_tags() {
718 let msg = Message::with_tags("test_topic", "tag1", b"test_body");
719 assert_eq!(msg.topic().as_str(), "test_topic");
720 assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
721 }
722
723 #[test]
724 fn test_message_with_tags_bytes() {
725 let body = Bytes::from_static(b"test_body");
726 let msg = Message::with_tags_bytes("test_topic", "tag1", body.clone());
727 assert_eq!(msg.topic().as_str(), "test_topic");
728 assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
729 assert_eq!(msg.body().unwrap(), body);
730 }
731
732 #[test]
733 fn test_message_with_keys() {
734 let msg = Message::with_keys("test_topic", "tag1", "key1", b"test_body");
735 assert_eq!(msg.topic().as_str(), "test_topic");
736 assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
737 assert_eq!(
738 msg.properties()
739 .as_map()
740 .get(MessageConst::PROPERTY_KEYS)
741 .unwrap()
742 .as_str(),
743 "key1"
744 );
745 }
746
747 #[test]
748 fn test_message_with_keys_bytes() {
749 let body = Bytes::from_static(b"test_body");
750 let msg = Message::with_keys_bytes("test_topic", "tag1", "key1", body.clone());
751 assert_eq!(msg.topic().as_str(), "test_topic");
752 assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
753 assert_eq!(msg.body().unwrap(), body);
754 }
755
756 #[test]
757 fn test_message_with_details() {
758 let msg = Message::with_details("test_topic", "tag1", "key1", 0, b"test_body", true);
759 assert_eq!(msg.topic().as_str(), "test_topic");
760 assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
761 assert!(msg.is_wait_store_msg_ok());
762 assert_eq!(
763 msg.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
764 Some(&CheetahString::from_static_str("true"))
765 );
766 }
767
768 #[test]
769 fn test_message_with_details_bytes() {
770 let body = Bytes::from_static(b"test_body");
771 let msg = Message::with_details_bytes("test_topic", "tag1", "key1", 0, body.clone(), false);
772 assert_eq!(msg.topic().as_str(), "test_topic");
773 assert!(!msg.is_wait_store_msg_ok());
774 assert_eq!(
775 msg.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
776 Some(&CheetahString::from_static_str("false"))
777 );
778 assert_eq!(msg.body().unwrap(), body);
779 }
780
781 #[test]
782 fn test_message_priority_matches_java_semantics() {
783 let mut msg = Message::new("test_topic", b"test_body");
784 assert_eq!(msg.priority(), -1);
785
786 msg.try_set_priority(3).unwrap();
787 assert_eq!(msg.priority(), 3);
788 assert_eq!(
789 msg.properties().as_map().get(MessageConst::PROPERTY_PRIORITY),
790 Some(&CheetahString::from_static_str("3"))
791 );
792
793 let error = msg
794 .try_set_priority(-1)
795 .expect_err("negative priority should be rejected");
796 assert_eq!(error.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
797 }
798
799 #[test]
800 fn test_get_transaction_id_returns_cheetah_string_ref() {
801 let mut msg = Message::new("test_topic", b"test_body");
802 msg.set_transaction_id(CheetahString::from_static_str("tx-123"));
803
804 let transaction_id = msg.get_transaction_id();
805
806 assert_eq!(transaction_id, Some(&CheetahString::from_static_str("tx-123")));
807 }
808
809 #[test]
810 fn message_display_redacts_body_properties_and_transaction_id() {
811 let mut msg = Message::with_tags("TopicA", "TagA", &[1, 255]);
812 msg.set_flag(7);
813 msg.set_transaction_id(CheetahString::from_static_str("tx-1"));
814
815 assert_eq!(
816 msg.to_string(),
817 "Message{topic='TopicA', flag=7, bodyLen=2, propertyCount=2, transactionIdPresent=true}"
818 );
819 }
820
821 #[test]
822 fn empty_message_display_contains_only_safe_metadata() {
823 let msg = Message::default();
824
825 assert_eq!(
826 msg.to_string(),
827 "Message{topic='', flag=0, bodyLen=0, propertyCount=0, transactionIdPresent=false}"
828 );
829 }
830
831 #[test]
832 fn test_properties_capacity_optimization() {
833 let msg1 = Message::with_details_bytes(
835 "topic",
836 CheetahString::empty(),
837 CheetahString::empty(),
838 0,
839 Bytes::from_static(b"body"),
840 true,
841 );
842 assert_eq!(msg1.properties().len(), 1);
843 assert_eq!(
844 msg1.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
845 Some(&CheetahString::from_static_str("true"))
846 );
847
848 let msg2 = Message::with_details_bytes(
850 "topic",
851 "tag1",
852 CheetahString::empty(),
853 0,
854 Bytes::from_static(b"body"),
855 true,
856 );
857 assert_eq!(msg2.properties().len(), 2);
858
859 let msg3 = Message::with_details_bytes("topic", "tag1", "key1", 0, Bytes::from_static(b"body"), true);
861 assert_eq!(msg3.properties().len(), 3);
862
863 let msg4 = Message::with_details_bytes("topic", "tag1", "key1", 0, Bytes::from_static(b"body"), false);
865 assert_eq!(msg4.properties().len(), 3);
866 assert_eq!(
867 msg4.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
868 Some(&CheetahString::from_static_str("false"))
869 );
870 }
871
872 #[test]
873 fn test_zero_copy_bytes() {
874 let original_bytes = Bytes::from_static(b"test_data");
875 let bytes_clone = original_bytes.clone();
876
877 let msg = Message::new_with_bytes("topic", bytes_clone);
879
880 let body = msg.body().unwrap();
882 assert_eq!(body.as_ptr(), original_bytes.as_ptr());
883 }
884
885 #[test]
886 fn test_put_user_property_error_handling() {
887 let mut msg = Message::new("test_topic", b"test body");
888
889 let result = msg.put_user_property(CheetahString::empty(), CheetahString::from_slice("value"));
891 assert_eq!(
892 result.expect_err("empty property name must fail").descriptor(),
893 &rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
894 );
895
896 let result = msg.put_user_property(CheetahString::from_slice("name"), CheetahString::empty());
898 assert_eq!(
899 result.expect_err("empty property value must fail").descriptor(),
900 &rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
901 );
902
903 let result = msg.put_user_property(CheetahString::from_slice("KEYS"), CheetahString::from_slice("value"));
905 assert_eq!(
906 result.expect_err("reserved property must fail").descriptor(),
907 &rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
908 );
909
910 let result = msg.put_user_property(
912 CheetahString::from_static_str(MessageConst::PROPERTY_PRIORITY),
913 CheetahString::from_slice("value"),
914 );
915 assert_eq!(
916 result.expect_err("priority property must fail").descriptor(),
917 &rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
918 );
919
920 let result = msg.put_user_property(
922 CheetahString::from_static_str(MessageConst::PROPERTY_ORIGIN_GROUP),
923 CheetahString::from_slice("value"),
924 );
925 assert_eq!(
926 result.expect_err("origin group property must fail").descriptor(),
927 &rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
928 );
929
930 let result = msg.put_user_property(
932 CheetahString::from_slice("my_custom_key"),
933 CheetahString::from_slice("my_value"),
934 );
935 assert!(result.is_ok());
936 assert_eq!(
937 msg.get_user_property(CheetahString::from_slice("my_custom_key"))
938 .unwrap()
939 .as_str(),
940 "my_value"
941 );
942 }
943}