rocketmq_common/common/message/

message_single.rs

// Copyright 2023 The RocketMQ Rust Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::any::Any;
use std::collections::HashMap;
use std::fmt;
use std::fmt::Display;
use std::fmt::Formatter;
use std::hash::DefaultHasher;
use std::hash::Hash;
use std::hash::Hasher;
use std::net::SocketAddr;

use bytes::Buf;
use bytes::BufMut;
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_ext::MessageExt;
use crate::common::message::message_flag::MessageFlag;
use crate::common::message::message_property::MessageProperties;
use crate::common::message::message_property::MessagePropertyKey;
use crate::common::message::MessageConst;
use crate::common::message::MessageTrait;
use crate::common::message::MessageVersion;
use crate::common::sys_flag::message_sys_flag::MessageSysFlag;
use crate::common::TopicFilterType;
use crate::MessageUtils;

#[derive(Clone, Debug)]
pub struct Message {
    topic: CheetahString,
    flag: MessageFlag,
    properties: MessageProperties,
    body: MessageBody,
    transaction_id: Option<CheetahString>,
}

// Private internal constructor used by builder
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 {
    /// Creates a new message builder.
    ///
    /// This is the recommended way to create messages.
    ///
    /// # Examples
    ///
    /// ```
    /// use rocketmq_common::common::message::message_single::Message;
    ///
    /// let msg = Message::builder()
    ///     .topic("test-topic")
    ///     .body_slice(b"hello world")
    ///     .tags("important")
    ///     .build_unchecked();
    /// ```
    pub fn builder() -> MessageBuilder {
        MessageBuilder::new()
    }

    /// Create a new message with topic and body slice (will copy)
    #[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)
    }

    /// Create a new message with topic and Bytes (zero-copy)
    #[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)
    }

    /// Create a new message with topic and Vec<u8> (zero-copy conversion to Bytes)
    #[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)
    }

    /// Create a message with tags and body slice (will copy)
    #[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)
    }

    /// Create a message with tags and Bytes (zero-copy)
    #[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)
    }

    /// Create a message with keys and body slice (will copy)
    #[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)
    }

    /// Create a message with keys and Bytes (zero-copy)
    #[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)
    }

    /// Create message with body slice (will copy data)
    #[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)
    }

    /// Create message with Bytes (zero-copy)
    #[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();

        // Pre-allocate HashMap with estimated capacity to avoid reallocation
        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);

        // Use static strings for keys to avoid allocations
        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();

        // Pre-allocate HashMap with estimated capacity
        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]
    fn set_wait_store_msg_ok(&mut self, wait_store_msg_ok: bool) {
        self.properties.as_map_mut().insert(
            CheetahString::from_static_str(MessageConst::PROPERTY_WAIT_STORE_MSG_OK),
            CheetahString::from_static_str(if wait_store_msg_ok { "true" } else { "false" }),
        );
    }

    #[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(),
        );
    }

    // ===== New Rust-idiomatic API methods =====

    /// Returns the message body as a byte slice (borrows).
    ///
    /// This is the recommended way to access the message body.
    #[inline]
    pub fn body_slice(&self) -> &[u8] {
        self.body.as_slice()
    }

    /// Consumes the message and returns the body.
    #[inline]
    pub fn into_body(self) -> MessageBody {
        self.body
    }

    /// Returns the message tags.
    #[inline]
    pub fn tags(&self) -> Option<&str> {
        self.properties.tags()
    }

    /// Returns the message keys as a vector.
    #[inline]
    pub fn keys(&self) -> Option<Vec<String>> {
        self.properties.keys()
    }

    /// Returns a property value by key.
    #[inline]
    pub fn property(&self, key: &str) -> Option<&str> {
        self.properties.as_map().get(key).map(|s| s.as_str())
    }

    /// Returns the message flag as a type-safe MessageFlag.
    #[inline]
    pub fn message_flag(&self) -> MessageFlag {
        self.flag
    }

    /// Returns whether to wait for store confirmation.
    #[inline]
    pub fn wait_store_msg_ok(&self) -> bool {
        self.is_wait_store_msg_ok()
    }

    /// Returns the delay time level.
    #[inline]
    pub fn delay_level(&self) -> i32 {
        self.delay_time_level()
    }

    /// Returns the buyer ID.
    #[inline]
    pub fn buyer_id(&self) -> Option<&str> {
        self.properties.buyer_id()
    }

    /// Returns the message priority, or `-1` if not set or not parseable.
    #[inline]
    pub fn priority(&self) -> i32 {
        self.properties.priority()
    }

    /// Sets the message priority with Java-compatible non-negative validation.
    #[inline]
    pub fn try_set_priority(&mut self, priority: i32) -> rocketmq_error::RocketMQResult<()> {
        if priority < 0 {
            return Err(rocketmq_error::RocketMQError::illegal_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(())
    }

    /// Returns the instance ID.
    #[inline]
    pub fn instance_id(&self) -> Option<&str> {
        self.properties.instance_id()
    }

    // ===== Internal accessors for other modules =====

    /// Returns a mutable reference to the topic (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn topic_mut(&mut self) -> &mut CheetahString {
        &mut self.topic
    }

    /// Returns a mutable reference to the flag (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn flag_mut(&mut self) -> &mut MessageFlag {
        &mut self.flag
    }

    /// Returns a mutable reference to properties (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn properties_mut(&mut self) -> &mut MessageProperties {
        &mut self.properties
    }

    /// Returns a mutable reference to the body (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn body_mut(&mut self) -> &mut MessageBody {
        &mut self.body
    }

    /// Returns a mutable reference to the transaction ID (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn transaction_id_mut(&mut self) -> &mut Option<CheetahString> {
        &mut self.transaction_id
    }

    /// Sets the topic (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn set_topic(&mut self, topic: CheetahString) {
        self.topic = topic;
    }

    /// Sets the flag (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn set_flag(&mut self, flag: i32) {
        self.flag = MessageFlag::from_bits(flag);
    }

    /// Sets the body (internal use only).
    #[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()
        };
    }

    /// Takes ownership of the body, leaving empty (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn take_body(&mut self) -> Option<Bytes> {
        let old = std::mem::take(&mut self.body);
        old.raw().cloned()
    }

    /// Returns a reference to the compressed body (internal use only).
    #[doc(hidden)]
    #[inline]
    pub fn compressed_body(&self) -> Option<&Bytes> {
        self.body.compressed()
    }
}

fn java_properties_to_string(properties: &HashMap<CheetahString, CheetahString>) -> String {
    if properties.is_empty() {
        return "{}".to_string();
    }

    let mut entries = properties
        .iter()
        .map(|(key, value)| (key.as_str(), value.as_str()))
        .collect::<Vec<_>>();
    entries.sort_unstable_by(|left, right| left.0.cmp(right.0).then_with(|| left.1.cmp(right.1)));

    let body = entries
        .into_iter()
        .map(|(key, value)| format!("{key}={value}"))
        .collect::<Vec<_>>()
        .join(", ");
    format!("{{{body}}}")
}

fn java_body_to_string(body: Option<&Bytes>) -> String {
    match body {
        Some(body) => {
            let values = body
                .iter()
                .map(|byte| (*byte as i8).to_string())
                .collect::<Vec<_>>()
                .join(", ");
            format!("[{values}]")
        }
        None => "null".to_string(),
    }
}

impl Display for Message {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "Message{{topic='{}', flag={}, properties={}, body={}, transactionId='{}'}}",
            self.topic,
            self.flag.bits(),
            java_properties_to_string(self.properties.as_map()),
            java_body_to_string(self.body.raw()),
            self.transaction_id.as_deref().unwrap_or("null")
        )
    }
}

#[allow(unused_variables)]
impl MessageTrait for Message {
    #[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!(error.to_string().contains("greater than or equal to 0"));
    }

    #[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_matches_java_to_string_shape() {
        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, properties={TAGS=TagA, WAIT=true}, body=[1, -1], transactionId='tx-1'}"
        );
    }

    #[test]
    fn empty_message_display_uses_java_array_null_wording_for_absent_body() {
        let msg = Message::default();

        assert_eq!(
            msg.to_string(),
            "Message{topic='', flag=0, properties={}, body=null, transactionId='null'}"
        );
    }

    #[test]
    fn test_properties_capacity_optimization() {
        // Test with no tags or keys
        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"))
        );

        // Test with tags
        let msg2 = Message::with_details_bytes(
            "topic",
            "tag1",
            CheetahString::empty(),
            0,
            Bytes::from_static(b"body"),
            true,
        );
        assert_eq!(msg2.properties().len(), 2);

        // Test with tags and keys
        let msg3 = Message::with_details_bytes("topic", "tag1", "key1", 0, Bytes::from_static(b"body"), true);
        assert_eq!(msg3.properties().len(), 3);

        // Test with wait_store_msg_ok = false
        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();

        // Creating message with Bytes should not copy
        let msg = Message::new_with_bytes("topic", bytes_clone);

        // The body should share the same underlying data
        let body = msg.body().unwrap();
        assert_eq!(body.as_ptr(), original_bytes.as_ptr());
    }

    #[test]
    fn test_put_user_property_error_handling() {
        use rocketmq_error::RocketMQError;

        let mut msg = Message::new("test_topic", b"test body");

        // Test empty name
        let result = msg.put_user_property(CheetahString::empty(), CheetahString::from_slice("value"));
        assert!(result.is_err());
        if let Err(RocketMQError::InvalidProperty(e)) = result {
            assert!(e.contains("null or blank"));
        }

        // Test empty value
        let result = msg.put_user_property(CheetahString::from_slice("name"), CheetahString::empty());
        assert!(result.is_err());
        if let Err(RocketMQError::InvalidProperty(e)) = result {
            assert!(e.contains("null or blank"));
        }

        // Test system reserved property
        let result = msg.put_user_property(CheetahString::from_slice("KEYS"), CheetahString::from_slice("value"));
        assert!(result.is_err());
        if let Err(RocketMQError::InvalidProperty(e)) = result {
            assert!(e.contains("used by system"));
        }

        // Test Java-reserved priority property
        let result = msg.put_user_property(
            CheetahString::from_static_str(MessageConst::PROPERTY_PRIORITY),
            CheetahString::from_slice("value"),
        );
        assert!(result.is_err());
        if let Err(RocketMQError::InvalidProperty(e)) = result {
            assert!(e.contains("used by system"));
        }

        // Test Java-reserved origin group property
        let result = msg.put_user_property(
            CheetahString::from_static_str(MessageConst::PROPERTY_ORIGIN_GROUP),
            CheetahString::from_slice("value"),
        );
        assert!(result.is_err());
        if let Err(RocketMQError::InvalidProperty(e)) = result {
            assert!(e.contains("used by system"));
        }

        // Test valid user property
        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"
        );
    }
}