[ISSUE #1223]Change the return type of the get_transaction_id method in MessageTrait to &CheetahString (#1224)

Author: mxsm

rocketmq-broker/src/processor/end_transaction_processor.rs

@@ -429,14 +429,14 @@ mod tests {
             msg_ext.reconsume_times
         );
         assert!(msg_inner.is_wait_store_msg_ok());
-        /*        assert_eq!(
+        assert_eq!(
             msg_inner.get_transaction_id(),
-            &msg_ext
+            msg_ext
                 .get_user_property(&CheetahString::from_static_str(
                     MessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX
                 ))
-                .unwrap_or_default()
-        );*/
+                .as_ref()
+        );
         assert_eq!(msg_inner.message_ext_inner.sys_flag, msg_ext.sys_flag);
         /*        assert_eq!(
             msg_inner.tags_code,

rocketmq-broker/src/transaction/queue/transactional_message_util.rs

@@ -79,13 +79,12 @@ impl TransactionalMessageUtil {
             .message_ext_inner
             .set_born_timestamp(msg_ext.born_timestamp);
         msg_inner.message_ext_inner.set_born_host(msg_ext.born_host);
-        msg_inner.set_transaction_id(
-            msg_ext
-                .get_property(&CheetahString::from_static_str(
-                    MessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
-                ))
-                .unwrap_or_default(),
-        );
+
+        if let Some(transaction_id) = msg_ext.get_property(&CheetahString::from_static_str(
+            MessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
+        )) {
+            msg_inner.set_transaction_id(transaction_id);
+        }
 
         MessageAccessor::set_properties(&mut msg_inner, msg_ext.get_properties().clone());
         MessageAccessor::put_property(
@@ -168,11 +167,11 @@ mod tests {
         assert_eq!(msg_inner.get_flag(), msg_ext.get_flag());
         assert_eq!(
             msg_inner.get_transaction_id(),
-            &msg_ext
+            msg_ext
                 .get_property(&CheetahString::from_static_str(
                     MessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX
                 ))
-                .unwrap_or_default()
+                .as_ref()
         );
     }
 

rocketmq-client/examples/transaction/transaction_producer.rs

@@ -92,13 +92,18 @@ impl TransactionListener for TransactionListenerImpl {
             .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
         let status = value % 3;
         let mut guard = self.local_trans.lock();
-        guard.insert(msg.get_transaction_id().clone(), status);
+        guard.insert(
+            msg.get_transaction_id().cloned().unwrap_or_default(),
+            status,
+        );
         LocalTransactionState::Unknown
     }
 
     fn check_local_transaction(&self, msg: &MessageExt) -> LocalTransactionState {
         let mut guard = self.local_trans.lock();
-        let status = guard.get(msg.get_transaction_id()).unwrap_or(&-1);
+        let status = guard
+            .get(&msg.get_transaction_id().cloned().unwrap_or_default())
+            .unwrap_or(&-1);
         match status {
             1 => LocalTransactionState::CommitMessage,
             2 => LocalTransactionState::RollbackMessage,

rocketmq-client/src/producer/producer_impl/default_mq_producer_impl.rs

@@ -2029,7 +2029,7 @@ impl DefaultMQProducerImpl {
             producer_group: self.producer_config.producer_group().clone(),
             message: msg,
             msg_id: msg_id.clone(),
-            transaction_id: msg.get_transaction_id().clone(),
+            transaction_id: msg.get_transaction_id().cloned().unwrap_or_default(),
             broker_addr: broker_addr.clone(),
             from_transaction_check,
             transaction_state: local_transaction_state,

rocketmq-common/src/common/message.rs

@@ -302,7 +302,7 @@ pub trait MessageTrait: Any + Display + Debug {
     /// # Returns
     ///
     /// A reference to the transaction ID as a `&str`.
-    fn get_transaction_id(&self) -> &CheetahString;
+    fn get_transaction_id(&self) -> Option<&CheetahString>;
 
     /// Sets the transaction ID for the message.
     ///

rocketmq-common/src/common/message/message_batch.rs

@@ -179,11 +179,8 @@ impl MessageTrait for MessageBatch {
         self.final_message.properties = properties;
     }
 
-    fn get_transaction_id(&self) -> &CheetahString {
-        self.final_message
-            .transaction_id
-            .as_ref()
-            .expect("transaction_id is None")
+    fn get_transaction_id(&self) -> Option<&CheetahString> {
+        self.final_message.transaction_id.as_ref()
     }
 
     fn set_transaction_id(&mut self, transaction_id: CheetahString) {

rocketmq-common/src/common/message/message_client_ext.rs

@@ -105,7 +105,7 @@ impl MessageTrait for MessageClientExt {
         self.message_ext_inner.set_properties(properties);
     }
 
-    fn get_transaction_id(&self) -> &CheetahString {
+    fn get_transaction_id(&self) -> Option<&CheetahString> {
         self.message_ext_inner.get_transaction_id()
     }
 

rocketmq-common/src/common/message/message_ext.rs

@@ -230,7 +230,7 @@ impl MessageExt {
 impl Default for MessageExt {
     fn default() -> Self {
         Self {
-            message: Default::default(),
+            message: Message::default(),
             broker_name: CheetahString::default(),
             queue_id: 0,
             store_size: 0,
@@ -321,7 +321,7 @@ impl MessageTrait for MessageExt {
     }
 
     #[inline]
-    fn get_transaction_id(&self) -> &CheetahString {
+    fn get_transaction_id(&self) -> Option<&CheetahString> {
         self.message.get_transaction_id()
     }
 

rocketmq-common/src/common/message/message_ext_broker_inner.rs

@@ -235,7 +235,7 @@ impl MessageTrait for MessageExtBrokerInner {
         self.message_ext_inner.set_properties(properties);
     }
 
-    fn get_transaction_id(&self) -> &CheetahString {
+    fn get_transaction_id(&self) -> Option<&CheetahString> {
         self.message_ext_inner.get_transaction_id()
     }
 

rocketmq-common/src/common/message/message_single.rs

@@ -38,7 +38,7 @@ use crate::common::sys_flag::message_sys_flag::MessageSysFlag;
 use crate::common::TopicFilterType;
 use crate::MessageUtils;
 
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug)]
 pub struct Message {
     pub topic: CheetahString,
     pub flag: i32,
@@ -50,6 +50,19 @@ pub struct Message {
     pub transaction_id: Option<CheetahString>,
 }
 
+impl Default for Message {
+    fn default() -> Self {
+        Self {
+            topic: CheetahString::new(),
+            flag: 0,
+            properties: HashMap::new(),
+            body: None,
+            compressed_body: None,
+            transaction_id: None,
+        }
+    }
+}
+
 impl Message {
     pub fn new(topic: impl Into<CheetahString>, body: &[u8]) -> Self {
         Self::with_details(
@@ -328,10 +341,8 @@ impl MessageTrait for Message {
     }
 
     #[inline]
-    fn get_transaction_id(&self) -> &CheetahString {
-        self.transaction_id
-            .as_ref()
-            .expect("transaction_id is None")
+    fn get_transaction_id(&self) -> Option<&CheetahString> {
+        self.transaction_id.as_ref()
     }
 
     fn set_transaction_id(&mut self, transaction_id: CheetahString) {