[ISSUE #1697]🚀EscapeBridge supports putMessageToRemoteBroker🔥 (#1708)

Author: mxsm

rocketmq-broker/src/failover/escape_bridge.rs

@@ -21,15 +21,20 @@ use rocketmq_client_rust::producer::send_result::SendResult;
 use rocketmq_client_rust::producer::send_status::SendStatus;
 use rocketmq_common::common::broker::broker_config::BrokerConfig;
 use rocketmq_common::common::message::message_ext_broker_inner::MessageExtBrokerInner;
+use rocketmq_common::common::message::message_queue::MessageQueue;
+use rocketmq_common::common::message::MessageConst;
 use rocketmq_common::common::message::MessageTrait;
 use rocketmq_common::common::mix_all;
 use rocketmq_runtime::RocketMQRuntime;
 use rocketmq_rust::ArcMut;
 use rocketmq_store::base::message_result::PutMessageResult;
 use rocketmq_store::base::message_status_enum::PutMessageStatus;
 use rocketmq_store::log_file::MessageStore;
+use tracing::warn;
 
+use crate::out_api::broker_outer_api::BrokerOuterAPI;
 use crate::topic::manager::topic_route_info_manager::TopicRouteInfoManager;
+use crate::transaction::queue::transactional_message_util::TransactionalMessageUtil;
 
 const SEND_TIMEOUT: u64 = 3_000;
 const DEFAULT_PULL_TIMEOUT_MILLIS: u64 = 10_000;
@@ -79,6 +84,7 @@ pub(crate) struct EscapeBridge<MS> {
     message_store: ArcMut<MS>,
     broker_config: Arc<BrokerConfig>,
     topic_route_info_manager: Arc<TopicRouteInfoManager>,
+    broker_outer_api: Arc<BrokerOuterAPI>,
 }
 
 impl<MS> EscapeBridge<MS>
@@ -104,10 +110,110 @@ where
 
     pub async fn put_message_to_remote_broker(
         &mut self,
-        _message_ext: MessageExtBrokerInner,
-        _broker_name_to_send: Option<CheetahString>,
+        message_ext: MessageExtBrokerInner,
+        mut broker_name_to_send: Option<CheetahString>,
     ) -> Option<SendResult> {
-        unimplemented!("EscapeBridge putMessageToRemoteBroker")
+        if broker_name_to_send.is_some()
+            && self.broker_config.broker_identity.broker_name.as_str()
+                == broker_name_to_send.as_ref().unwrap().as_str()
+        {
+            return None;
+        }
+        let is_trans_half_message =
+            TransactionalMessageUtil::build_half_topic() == message_ext.get_topic();
+        let mut message_to_put = if is_trans_half_message {
+            TransactionalMessageUtil::build_transactional_message_from_half_message(
+                &message_ext.message_ext_inner,
+            )
+        } else {
+            message_ext
+        };
+        let topic_publish_info = self
+            .topic_route_info_manager
+            .try_to_find_topic_publish_info(message_to_put.get_topic())
+            .await;
+        if !topic_publish_info.as_ref().is_some_and(|value| value.ok()) {
+            warn!(
+                "putMessageToRemoteBroker: no route info of topic {} when escaping message, \
+                 msgId={}",
+                message_to_put.get_topic(),
+                message_to_put.message_ext_inner.msg_id
+            );
+            return None;
+        }
+        let topic_publish_info = topic_publish_info.unwrap();
+        let _mq_selected = if !broker_name_to_send
+            .as_ref()
+            .is_some_and(|value| !value.is_empty())
+        {
+            let mq = topic_publish_info
+                .select_one_message_queue_by_broker(broker_name_to_send.as_ref())
+                .unwrap();
+            message_to_put.message_ext_inner.queue_id = mq.get_queue_id();
+            broker_name_to_send = Some(mq.get_broker_name().clone());
+            if self.broker_config.broker_identity.broker_name.as_str()
+                == mq.get_broker_name().as_str()
+            {
+                warn!(
+                    "putMessageToRemoteBroker failed, remote broker not found. Topic: {}, MsgId: \
+                     {}, Broker: {}",
+                    message_to_put.get_topic(),
+                    message_to_put.message_ext_inner.msg_id,
+                    mq.get_broker_name()
+                );
+                return None;
+            }
+            mq
+        } else {
+            MessageQueue::from_parts(
+                message_to_put.get_topic(),
+                broker_name_to_send.clone().unwrap(),
+                message_to_put.queue_id(),
+            )
+        };
+        let broker_addr_to_send = self
+            .topic_route_info_manager
+            .find_broker_address_in_publish(broker_name_to_send.as_ref());
+        if broker_addr_to_send.is_none() {
+            warn!(
+                "putMessageToRemoteBroker failed, remote broker not found. Topic: {}, MsgId: {}, \
+                 Broker: {}",
+                message_to_put.get_topic(),
+                message_to_put.message_ext_inner.msg_id,
+                broker_name_to_send.as_ref().unwrap()
+            );
+            return None;
+        }
+        let producer_group = self.get_producer_group(&message_to_put);
+        let result = self
+            .broker_outer_api
+            .send_message_to_specific_broker(
+                broker_addr_to_send.as_ref().unwrap(),
+                message_to_put.message_ext_inner,
+                producer_group,
+                SEND_TIMEOUT,
+            )
+            .await;
+        match result {
+            Ok(value) => {
+                if value.send_status == SendStatus::SendOk {
+                    Some(value)
+                } else {
+                    None
+                }
+            }
+            Err(_) => None,
+        }
+    }
+
+    fn get_producer_group(&self, message_ext: &MessageExtBrokerInner) -> CheetahString {
+        let producer_group = message_ext.get_property(&CheetahString::from_static_str(
+            MessageConst::PROPERTY_PRODUCER_GROUP,
+        ));
+        match producer_group {
+            None => self.inner_producer_group_name.clone(),
+            Some(value) => value,
+        }
     }
 }
 

rocketmq-broker/src/out_api/broker_outer_api.rs

@@ -20,8 +20,10 @@ use std::sync::Weak;
 
 use cheetah_string::CheetahString;
 use dns_lookup::lookup_host;
+use rocketmq_client_rust::producer::send_result::SendResult;
 use rocketmq_common::common::broker::broker_config::BrokerIdentity;
 use rocketmq_common::common::config::TopicConfig;
+use rocketmq_common::common::message::message_ext::MessageExt;
 use rocketmq_common::common::message::message_queue::MessageQueue;
 use rocketmq_common::utils::crc32_utils;
 use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
@@ -452,6 +454,16 @@ impl BrokerOuterAPI {
             "".to_string(),
         ))
     }
+
+    pub async fn send_message_to_specific_broker(
+        &self,
+        _broker_addr: &CheetahString,
+        _msg: MessageExt,
+        _group: CheetahString,
+        _timeout_millis: u64,
+    ) -> Result<SendResult> {
+        unimplemented!("sendMessageToSpecificBroker")
+    }
 }
 
 fn dns_lookup_address_by_domain(domain: &str) -> Vec<CheetahString> {