use std::any::Any;
use std::collections::HashMap;
use std::fmt;
use std::fmt::Display;
use std::fmt::Formatter;
use bytes::Bytes;
use cheetah_string::CheetahString;
use crate::common::hasher::string_hasher::JavaStringHasher;
use crate::common::message::message_body::MessageBody;
use crate::common::message::message_builder::MessageBuilder;
use crate::common::message::message_flag::MessageFlag;
use crate::common::message::message_property::MessageProperties;
use crate::common::message::MessageConst;
use crate::common::message::MessageTrait;
use crate::common::sys_flag::message_sys_flag::MessageSysFlag;
use crate::common::TopicFilterType;
#[derive(Clone)]
pub struct Message {
topic: CheetahString,
flag: MessageFlag,
properties: MessageProperties,
body: MessageBody,
transaction_id: Option<CheetahString>,
}
impl Message {
pub(crate) fn from_builder(
topic: CheetahString,
body: MessageBody,
properties: MessageProperties,
flag: MessageFlag,
transaction_id: Option<CheetahString>,
) -> Self {
Self {
topic,
flag,
properties,
body,
transaction_id,
}
}
}
impl Default for Message {
fn default() -> Self {
Self {
topic: CheetahString::new(),
flag: MessageFlag::empty(),
properties: MessageProperties::new(),
body: MessageBody::empty(),
transaction_id: None,
}
}
}
impl Message {
pub fn builder() -> MessageBuilder {
MessageBuilder::new()
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn new(topic: impl Into<CheetahString>, body: &[u8]) -> Self {
Self::with_details(topic, CheetahString::empty(), CheetahString::empty(), 0, body, true)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn new_with_bytes(topic: impl Into<CheetahString>, body: Bytes) -> Self {
Self::with_details_bytes(topic, CheetahString::empty(), CheetahString::empty(), 0, body, true)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn new_with_vec(topic: impl Into<CheetahString>, body: Vec<u8>) -> Self {
Self::with_details_bytes(
topic,
CheetahString::empty(),
CheetahString::empty(),
0,
Bytes::from(body),
true,
)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn new_body(topic: impl Into<CheetahString>, body: Option<Bytes>) -> Self {
Self::with_details_body(topic, CheetahString::empty(), CheetahString::empty(), 0, body, true)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn with_tags(topic: impl Into<CheetahString>, tags: impl Into<CheetahString>, body: &[u8]) -> Self {
Self::with_details(topic, tags, CheetahString::empty(), 0, body, true)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn with_tags_bytes(topic: impl Into<CheetahString>, tags: impl Into<CheetahString>, body: Bytes) -> Self {
Self::with_details_bytes(topic, tags, CheetahString::empty(), 0, body, true)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn with_keys(
topic: impl Into<CheetahString>,
tags: impl Into<CheetahString>,
keys: impl Into<CheetahString>,
body: &[u8],
) -> Self {
Self::with_details(topic, tags, keys, 0, body, true)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
#[inline]
pub fn with_keys_bytes(
topic: impl Into<CheetahString>,
tags: impl Into<CheetahString>,
keys: impl Into<CheetahString>,
body: Bytes,
) -> Self {
Self::with_details_bytes(topic, tags, keys, 0, body, true)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
#[allow(deprecated)]
pub fn with_details(
topic: impl Into<CheetahString>,
tags: impl Into<CheetahString>,
keys: impl Into<CheetahString>,
flag: i32,
body: &[u8],
wait_store_msg_ok: bool,
) -> Self {
Self::with_details_bytes(topic, tags, keys, flag, Bytes::copy_from_slice(body), wait_store_msg_ok)
}
#[deprecated(since = "0.8.0", note = "Use Message::builder() instead")]
pub fn with_details_bytes(
topic: impl Into<CheetahString>,
tags: impl Into<CheetahString>,
keys: impl Into<CheetahString>,
flag: i32,
body: Bytes,
wait_store_msg_ok: bool,
) -> Self {
let topic = topic.into();
let tags = tags.into();
let keys = keys.into();
let has_tags = !tags.is_empty();
let has_keys = !keys.is_empty();
let initial_capacity = (has_tags as usize) + (has_keys as usize) + 1;
let mut properties = HashMap::with_capacity(initial_capacity);
if has_tags {
properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_TAGS), tags);
}
if has_keys {
properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_KEYS), keys);
}
properties.insert(
CheetahString::from_static_str(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
CheetahString::from_static_str(if wait_store_msg_ok { "true" } else { "false" }),
);
Message {
topic,
flag: MessageFlag::from_bits(flag),
properties: MessageProperties::from_map(properties),
body: MessageBody::from(body),
transaction_id: None,
}
}
pub fn with_details_body(
topic: impl Into<CheetahString>,
tags: impl Into<CheetahString>,
keys: impl Into<CheetahString>,
flag: i32,
body: Option<Bytes>,
wait_store_msg_ok: bool,
) -> Self {
let topic = topic.into();
let tags = tags.into();
let keys = keys.into();
let has_tags = !tags.is_empty();
let has_keys = !keys.is_empty();
let initial_capacity = (has_tags as usize) + (has_keys as usize) + 1;
let mut properties = HashMap::with_capacity(initial_capacity);
if has_tags {
properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_TAGS), tags);
}
if has_keys {
properties.insert(CheetahString::from_static_str(MessageConst::PROPERTY_KEYS), keys);
}
properties.insert(
CheetahString::from_static_str(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
CheetahString::from_static_str(if wait_store_msg_ok { "true" } else { "false" }),
);
Message {
topic,
flag: MessageFlag::from_bits(flag),
properties: MessageProperties::from_map(properties),
body: if let Some(b) = body {
MessageBody::from(b)
} else {
MessageBody::empty()
},
transaction_id: None,
}
}
#[inline]
pub fn set_tags(&mut self, tags: CheetahString) {
self.properties
.as_map_mut()
.insert(CheetahString::from_static_str(MessageConst::PROPERTY_TAGS), tags);
}
#[inline]
pub fn set_keys(&mut self, keys: CheetahString) {
self.properties
.as_map_mut()
.insert(CheetahString::from_static_str(MessageConst::PROPERTY_KEYS), keys);
}
#[inline]
pub fn clear_property(&mut self, name: impl Into<CheetahString>) {
self.properties.as_map_mut().remove(name.into().as_str());
}
#[inline]
pub fn set_properties(&mut self, properties: HashMap<CheetahString, CheetahString>) {
self.properties = MessageProperties::from_map(properties);
}
#[inline]
pub fn get_property(&self, key: &CheetahString) -> Option<CheetahString> {
self.properties.as_map().get(key).cloned()
}
#[inline]
pub fn body(&self) -> Option<bytes::Bytes> {
self.body.raw().cloned()
}
#[inline]
pub fn flag(&self) -> i32 {
self.flag.bits()
}
#[inline]
pub fn topic(&self) -> &CheetahString {
&self.topic
}
#[inline]
pub fn properties(&self) -> &MessageProperties {
&self.properties
}
#[inline]
pub fn transaction_id(&self) -> Option<&str> {
self.transaction_id.as_deref()
}
#[inline]
pub fn get_transaction_id(&self) -> Option<&CheetahString> {
self.transaction_id.as_ref()
}
#[inline]
pub fn get_tags(&self) -> Option<CheetahString> {
self.properties.as_map().get(MessageConst::PROPERTY_TAGS).cloned()
}
#[inline]
pub fn is_wait_store_msg_ok(&self) -> bool {
self.properties.wait_store_msg_ok()
}
#[inline]
pub fn delay_time_level(&self) -> i32 {
self.properties.delay_level().unwrap_or(0)
}
#[inline]
pub fn set_delay_time_level(&mut self, level: i32) {
self.properties.as_map_mut().insert(
CheetahString::from_static_str(MessageConst::PROPERTY_DELAY_TIME_LEVEL),
CheetahString::from(level.to_string()),
);
}
#[inline]
pub fn get_user_property(&self, name: impl Into<CheetahString>) -> Option<CheetahString> {
self.properties.as_map().get(name.into().as_str()).cloned()
}
#[inline]
pub fn as_any(&self) -> &dyn Any {
self
}
#[inline]
pub fn set_instance_id(&mut self, instance_id: impl Into<CheetahString>) {
self.properties.as_map_mut().insert(
CheetahString::from_static_str(MessageConst::PROPERTY_INSTANCE_ID),
instance_id.into(),
);
}
#[inline]
pub fn body_slice(&self) -> &[u8] {
self.body.as_slice()
}
#[inline]
pub fn into_body(self) -> MessageBody {
self.body
}
#[inline]
pub fn tags(&self) -> Option<&str> {
self.properties.tags()
}
#[inline]
pub fn keys(&self) -> Option<Vec<String>> {
self.properties.keys()
}
#[inline]
pub fn property(&self, key: &str) -> Option<&str> {
self.properties.as_map().get(key).map(|s| s.as_str())
}
#[inline]
pub fn message_flag(&self) -> MessageFlag {
self.flag
}
#[inline]
pub fn wait_store_msg_ok(&self) -> bool {
self.is_wait_store_msg_ok()
}
#[inline]
pub fn delay_level(&self) -> i32 {
self.delay_time_level()
}
#[inline]
pub fn buyer_id(&self) -> Option<&str> {
self.properties.buyer_id()
}
#[inline]
pub fn priority(&self) -> i32 {
self.properties.priority()
}
#[inline]
pub fn try_set_priority(&mut self, priority: i32) -> rocketmq_error::Result<()> {
if priority < 0 {
return Err(crate::error::invalid_argument(
"The priority must be greater than or equal to 0",
));
}
self.properties.as_map_mut().insert(
CheetahString::from_static_str(MessageConst::PROPERTY_PRIORITY),
CheetahString::from(priority.to_string()),
);
Ok(())
}
#[inline]
pub fn instance_id(&self) -> Option<&str> {
self.properties.instance_id()
}
#[doc(hidden)]
#[inline]
pub fn topic_mut(&mut self) -> &mut CheetahString {
&mut self.topic
}
#[doc(hidden)]
#[inline]
pub fn flag_mut(&mut self) -> &mut MessageFlag {
&mut self.flag
}
#[doc(hidden)]
#[inline]
pub fn properties_mut(&mut self) -> &mut MessageProperties {
&mut self.properties
}
#[doc(hidden)]
#[inline]
pub fn body_mut(&mut self) -> &mut MessageBody {
&mut self.body
}
#[doc(hidden)]
#[inline]
pub fn transaction_id_mut(&mut self) -> &mut Option<CheetahString> {
&mut self.transaction_id
}
#[doc(hidden)]
#[inline]
pub fn set_topic(&mut self, topic: CheetahString) {
self.topic = topic;
}
#[doc(hidden)]
#[inline]
pub fn set_flag(&mut self, flag: i32) {
self.flag = MessageFlag::from_bits(flag);
}
#[doc(hidden)]
#[inline]
pub fn set_body(&mut self, body: Option<Bytes>) {
self.body = if let Some(b) = body {
MessageBody::from(b)
} else {
MessageBody::empty()
};
}
#[doc(hidden)]
#[inline]
pub fn take_body(&mut self) -> Option<Bytes> {
let old = std::mem::take(&mut self.body);
old.raw().cloned()
}
#[doc(hidden)]
#[inline]
pub fn compressed_body(&self) -> Option<&Bytes> {
self.body.compressed()
}
}
impl fmt::Debug for Message {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
f.debug_struct("Message")
.field("topic", &self.topic)
.field("flag", &self.flag.bits())
.field("body_len", &self.body_slice().len())
.field("property_count", &self.properties.len())
.field("transaction_id_present", &self.transaction_id.is_some())
.finish()
}
}
impl Display for Message {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"Message{{topic='{}', flag={}, bodyLen={}, propertyCount={}, transactionIdPresent={}}}",
self.topic,
self.flag.bits(),
self.body_slice().len(),
self.properties.len(),
self.transaction_id.is_some()
)
}
}
#[allow(unused_variables)]
impl MessageTrait for Message {
#[inline]
fn into_message(self) -> Message {
self
}
#[inline]
fn put_property(&mut self, key: CheetahString, value: CheetahString) {
self.properties.as_map_mut().insert(key, value);
}
#[inline]
fn clear_property(&mut self, name: &str) {
self.properties.as_map_mut().remove(name);
}
#[inline]
fn property(&self, name: &CheetahString) -> Option<CheetahString> {
self.properties.as_map().get(name).cloned()
}
fn property_ref(&self, name: &CheetahString) -> Option<&CheetahString> {
self.properties.as_map().get(name)
}
#[inline]
fn topic(&self) -> &CheetahString {
&self.topic
}
#[inline]
fn set_topic(&mut self, topic: CheetahString) {
self.set_topic(topic);
}
#[inline]
fn get_flag(&self) -> i32 {
self.flag.bits()
}
#[inline]
fn set_flag(&mut self, flag: i32) {
self.set_flag(flag);
}
#[inline]
fn get_body(&self) -> Option<&Bytes> {
self.body.raw()
}
#[inline]
fn set_body(&mut self, body: Bytes) {
self.set_body(Some(body));
}
#[inline]
fn get_properties(&self) -> &HashMap<CheetahString, CheetahString> {
self.properties.as_map()
}
#[inline]
fn set_properties(&mut self, properties: HashMap<CheetahString, CheetahString>) {
self.properties = MessageProperties::from_map(properties);
}
#[inline]
fn transaction_id(&self) -> Option<&CheetahString> {
self.transaction_id.as_ref()
}
#[inline]
fn set_transaction_id(&mut self, transaction_id: CheetahString) {
*self.transaction_id_mut() = Some(transaction_id);
}
#[inline]
fn get_compressed_body_mut(&mut self) -> Option<&mut Bytes> {
self.body.compressed_mut().as_mut()
}
#[inline]
fn get_compressed_body(&self) -> Option<&Bytes> {
self.compressed_body()
}
#[inline]
fn set_compressed_body_mut(&mut self, compressed_body: Bytes) {
self.body.set_compressed(compressed_body);
}
#[inline]
fn take_body(&mut self) -> Option<Bytes> {
self.take_body()
}
#[inline]
fn as_any(&self) -> &dyn Any {
self
}
#[inline]
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
}
pub fn parse_topic_filter_type(sys_flag: i32) -> TopicFilterType {
if (sys_flag & MessageSysFlag::MULTI_TAGS_FLAG) == MessageSysFlag::MULTI_TAGS_FLAG {
TopicFilterType::MultiTag
} else {
TopicFilterType::SingleTag
}
}
pub fn tags_string2tags_code(tags: Option<&CheetahString>) -> i64 {
match tags {
Some(tags) if !tags.is_empty() => JavaStringHasher::hash_str(tags.as_str()) as i64,
_ => 0,
}
}
#[cfg(test)]
#[allow(deprecated)]
mod tests {
use bytes::Bytes;
use super::*;
#[test]
fn test_message_new() {
let msg = Message::new("test_topic", b"test_body");
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.body().unwrap().as_ref(), b"test_body");
}
#[test]
fn test_message_new_with_bytes() {
let body = Bytes::from_static(b"test_body");
let msg = Message::new_with_bytes("test_topic", body.clone());
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.body().unwrap(), body);
}
#[test]
fn test_message_new_with_vec() {
let body = vec![1u8, 2, 3, 4, 5];
let msg = Message::new_with_vec("test_topic", body.clone());
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.body().unwrap().as_ref(), body.as_slice());
}
#[test]
fn test_message_with_tags() {
let msg = Message::with_tags("test_topic", "tag1", b"test_body");
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
}
#[test]
fn test_message_with_tags_bytes() {
let body = Bytes::from_static(b"test_body");
let msg = Message::with_tags_bytes("test_topic", "tag1", body.clone());
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
assert_eq!(msg.body().unwrap(), body);
}
#[test]
fn test_message_with_keys() {
let msg = Message::with_keys("test_topic", "tag1", "key1", b"test_body");
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
assert_eq!(
msg.properties()
.as_map()
.get(MessageConst::PROPERTY_KEYS)
.unwrap()
.as_str(),
"key1"
);
}
#[test]
fn test_message_with_keys_bytes() {
let body = Bytes::from_static(b"test_body");
let msg = Message::with_keys_bytes("test_topic", "tag1", "key1", body.clone());
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
assert_eq!(msg.body().unwrap(), body);
}
#[test]
fn test_message_with_details() {
let msg = Message::with_details("test_topic", "tag1", "key1", 0, b"test_body", true);
assert_eq!(msg.topic().as_str(), "test_topic");
assert_eq!(msg.get_tags().unwrap().as_str(), "tag1");
assert!(msg.is_wait_store_msg_ok());
assert_eq!(
msg.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
Some(&CheetahString::from_static_str("true"))
);
}
#[test]
fn test_message_with_details_bytes() {
let body = Bytes::from_static(b"test_body");
let msg = Message::with_details_bytes("test_topic", "tag1", "key1", 0, body.clone(), false);
assert_eq!(msg.topic().as_str(), "test_topic");
assert!(!msg.is_wait_store_msg_ok());
assert_eq!(
msg.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
Some(&CheetahString::from_static_str("false"))
);
assert_eq!(msg.body().unwrap(), body);
}
#[test]
fn test_message_priority_matches_java_semantics() {
let mut msg = Message::new("test_topic", b"test_body");
assert_eq!(msg.priority(), -1);
msg.try_set_priority(3).unwrap();
assert_eq!(msg.priority(), 3);
assert_eq!(
msg.properties().as_map().get(MessageConst::PROPERTY_PRIORITY),
Some(&CheetahString::from_static_str("3"))
);
let error = msg
.try_set_priority(-1)
.expect_err("negative priority should be rejected");
assert_eq!(error.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
}
#[test]
fn test_get_transaction_id_returns_cheetah_string_ref() {
let mut msg = Message::new("test_topic", b"test_body");
msg.set_transaction_id(CheetahString::from_static_str("tx-123"));
let transaction_id = msg.get_transaction_id();
assert_eq!(transaction_id, Some(&CheetahString::from_static_str("tx-123")));
}
#[test]
fn message_display_redacts_body_properties_and_transaction_id() {
let mut msg = Message::with_tags("TopicA", "TagA", &[1, 255]);
msg.set_flag(7);
msg.set_transaction_id(CheetahString::from_static_str("tx-1"));
assert_eq!(
msg.to_string(),
"Message{topic='TopicA', flag=7, bodyLen=2, propertyCount=2, transactionIdPresent=true}"
);
}
#[test]
fn empty_message_display_contains_only_safe_metadata() {
let msg = Message::default();
assert_eq!(
msg.to_string(),
"Message{topic='', flag=0, bodyLen=0, propertyCount=0, transactionIdPresent=false}"
);
}
#[test]
fn test_properties_capacity_optimization() {
let msg1 = Message::with_details_bytes(
"topic",
CheetahString::empty(),
CheetahString::empty(),
0,
Bytes::from_static(b"body"),
true,
);
assert_eq!(msg1.properties().len(), 1);
assert_eq!(
msg1.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
Some(&CheetahString::from_static_str("true"))
);
let msg2 = Message::with_details_bytes(
"topic",
"tag1",
CheetahString::empty(),
0,
Bytes::from_static(b"body"),
true,
);
assert_eq!(msg2.properties().len(), 2);
let msg3 = Message::with_details_bytes("topic", "tag1", "key1", 0, Bytes::from_static(b"body"), true);
assert_eq!(msg3.properties().len(), 3);
let msg4 = Message::with_details_bytes("topic", "tag1", "key1", 0, Bytes::from_static(b"body"), false);
assert_eq!(msg4.properties().len(), 3);
assert_eq!(
msg4.properties().as_map().get(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
Some(&CheetahString::from_static_str("false"))
);
}
#[test]
fn test_zero_copy_bytes() {
let original_bytes = Bytes::from_static(b"test_data");
let bytes_clone = original_bytes.clone();
let msg = Message::new_with_bytes("topic", bytes_clone);
let body = msg.body().unwrap();
assert_eq!(body.as_ptr(), original_bytes.as_ptr());
}
#[test]
fn test_put_user_property_error_handling() {
let mut msg = Message::new("test_topic", b"test body");
let result = msg.put_user_property(CheetahString::empty(), CheetahString::from_slice("value"));
assert_eq!(
result.expect_err("empty property name must fail").descriptor(),
&rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
);
let result = msg.put_user_property(CheetahString::from_slice("name"), CheetahString::empty());
assert_eq!(
result.expect_err("empty property value must fail").descriptor(),
&rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
);
let result = msg.put_user_property(CheetahString::from_slice("KEYS"), CheetahString::from_slice("value"));
assert_eq!(
result.expect_err("reserved property must fail").descriptor(),
&rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
);
let result = msg.put_user_property(
CheetahString::from_static_str(MessageConst::PROPERTY_PRIORITY),
CheetahString::from_slice("value"),
);
assert_eq!(
result.expect_err("priority property must fail").descriptor(),
&rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
);
let result = msg.put_user_property(
CheetahString::from_static_str(MessageConst::PROPERTY_ORIGIN_GROUP),
CheetahString::from_slice("value"),
);
assert_eq!(
result.expect_err("origin group property must fail").descriptor(),
&rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID
);
let result = msg.put_user_property(
CheetahString::from_slice("my_custom_key"),
CheetahString::from_slice("my_value"),
);
assert!(result.is_ok());
assert_eq!(
msg.get_user_property(CheetahString::from_slice("my_custom_key"))
.unwrap()
.as_str(),
"my_value"
);
}
}