[ISSUE #262]🚀Support register borker(request code:103)

Author: mxsm

rocketmq-broker/src/broker_runtime.rs

@@ -243,8 +243,8 @@ impl BrokerRuntime {
             .unwrap()
             .get_handle()
             .spawn(async move {
-                let period = Duration::from_secs(5);
-                let initial_delay = Some(Duration::from_secs(5));
+                let period = Duration::from_secs(10);
+                let initial_delay = Some(Duration::from_secs(60));
                 // initial delay
                 if let Some(initial_delay_inner) = initial_delay {
                     tokio::time::sleep(initial_delay_inner).await;
@@ -272,10 +272,6 @@ impl BrokerRuntime {
 
     /// Register broker to name server
     pub(crate) async fn register_broker_all(
-        /* broker_config: Arc<BrokerConfig>,
-        topic_config_manager: Arc<RwLock<TopicConfigManager>>,
-        topic_queue_mapping_manager: Arc<RwLock<TopicQueueMappingManager>>,
-        broker_out_api: Arc<RwLock<BrokerOuterAPI>>,*/
         &mut self,
         check_order_config: bool,
         oneway: bool,
@@ -307,15 +303,8 @@ impl BrokerRuntime {
             let topic_config_wrapper = self
                 .topic_config_manager
                 .build_serialize_wrapper(topic_config_table.clone());
-            self.do_register_broker_all(
-                /*Self::do_register_broker_all(
-                self.broker_config.clone(),
-                self.broker_out_api.clone(),*/
-                check_order_config,
-                oneway,
-                topic_config_wrapper,
-            )
-            .await;
+            self.do_register_broker_all(check_order_config, oneway, topic_config_wrapper)
+                .await;
             topic_config_table.clear();
         }
 
@@ -359,15 +348,8 @@ impl BrokerRuntime {
                 self.broker_config.is_in_broker_container,
             )
         {
-            self.do_register_broker_all(
-                /*Self::do_register_broker_all(
-                self.broker_config.clone(),
-                self.broker_out_api.clone(),*/
-                check_order_config,
-                oneway,
-                topic_config_wrapper,
-            )
-            .await;
+            self.do_register_broker_all(check_order_config, oneway, topic_config_wrapper)
+                .await;
         }
     }
 
@@ -383,8 +365,6 @@ impl BrokerRuntime {
     }
 
     async fn do_register_broker_all(
-        /*broker_config: Arc<BrokerConfig>,
-        broker_out_api: Arc<RwLock<BrokerOuterAPI>>,*/
         &mut self,
         _check_order_config: bool,
         oneway: bool,

rocketmq-broker/src/out_api/broker_outer_api.rs

@@ -106,6 +106,8 @@ impl BrokerOuterAPI {
                 heartbeat_timeout_millis,
                 body_crc32: 0,
             };
+
+            //build request body
             let request_body = RegisterBrokerBody {
                 topic_config_serialize_wrapper: topic_config_wrapper,
                 filter_server_list,
@@ -119,15 +121,18 @@ impl BrokerOuterAPI {
                 let cloned_body = body.clone();
                 let cloned_header = request_header.clone();
                 let addr = namesrv_addr.clone();
-                let handle = self
-                    .register_broker(addr, oneway, timeout_mills, cloned_header, cloned_body)
-                    .await;
+                let handle =
+                    self.register_broker(addr, oneway, timeout_mills, cloned_header, cloned_body);
                 handle_vec.push(handle);
             }
 
-            for handle in handle_vec {
+            /*for handle in handle_vec {
                 //let result = self.broker_outer_executor.spawn(handle).await;
                 register_broker_result_list.push(handle.unwrap());
+            }*/
+            while let Some(handle) = handle_vec.pop() {
+                let result = tokio::join!(handle);
+                register_broker_result_list.push(result.0.unwrap());
             }
         }
 
@@ -147,10 +152,10 @@ impl BrokerOuterAPI {
                 .set_body(Some(body.clone()));
 
         if oneway {
-            // match remoting_client.lock().unwrap().invoke_oneway(
             match self
                 .remoting_client
                 .invoke_oneway(namesrv_addr, request, timeout_mills)
+                .await
             {
                 Ok(_) => return None,
                 Err(_) => {

rocketmq-remoting/src/clients.rs

@@ -102,7 +102,7 @@ pub trait RemotingClient: RemotingService {
         invoke_callback: impl InvokeCallback,
     ) -> Result<(), Box<dyn std::error::Error>>;
 
-    fn invoke_oneway(
+    async fn invoke_oneway(
         &self,
         addr: String,
         request: RemotingCommand,

rocketmq-remoting/src/clients/blocking_client.rs

@@ -62,10 +62,10 @@ impl BlockingClient {
         request: RemotingCommand,
         timeout: Duration,
     ) -> anyhow::Result<()> {
-        match self.rt.block_on(tokio::time::timeout(
-            timeout,
-            self.inner.send_request(request),
-        )) {
+        match self
+            .rt
+            .block_on(tokio::time::timeout(timeout, self.inner.send(request)))
+        {
             Ok(Ok(_)) => Ok(()),
             Ok(Err(err)) => Err(err.into()),
             Err(err) => Err(err.into()),

rocketmq-remoting/src/clients/client.rs

@@ -60,9 +60,9 @@ impl Client {
     ///
     /// The `RemotingCommand` representing the response, wrapped in a `Result`. Returns an error if
     /// the invocation fails.
-    pub async fn invoke(&mut self, request: RemotingCommand) -> anyhow::Result<RemotingCommand> {
-        self.send_request(request).await?;
-        let response = self.read_response().await?;
+    pub async fn send_read(&mut self, request: RemotingCommand) -> anyhow::Result<RemotingCommand> {
+        self.send(request).await?;
+        let response = self.read().await?;
         Ok(response)
     }
 
@@ -90,10 +90,7 @@ impl Client {
     /// # Returns
     ///
     /// A `Result` indicating success or failure in sending the request.
-    pub async fn send_request(
-        &mut self,
-        request: RemotingCommand,
-    ) -> anyhow::Result<(), RemotingError> {
+    pub async fn send(&mut self, request: RemotingCommand) -> anyhow::Result<(), RemotingError> {
         self.connection.framed.send(request).await?;
         Ok(())
     }
@@ -104,7 +101,7 @@ impl Client {
     ///
     /// The `RemotingCommand` representing the response, wrapped in a `Result`. Returns an error if
     /// reading the response fails.
-    async fn read_response(&mut self) -> anyhow::Result<RemotingCommand, RemotingError> {
+    async fn read(&mut self) -> anyhow::Result<RemotingCommand, RemotingError> {
         let response = self.connection.framed.next().await;
         response.unwrap()
     }

rocketmq-remoting/src/clients/rocketmq_default_impl.rs

@@ -14,23 +14,20 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-use std::{collections::HashMap, error::Error, sync::Arc};
+use std::{collections::HashMap, error::Error, sync::Arc, time::Duration};
 
 use rocketmq_common::TokioExecutorService;
+use tokio::{runtime::Handle, time, time::timeout};
 use tracing::info;
 
 use crate::{
     clients::{Client, RemotingClient},
     protocol::remoting_command::RemotingCommand,
     remoting::{InvokeCallback, RemotingService},
-    runtime::{
-        config::client_config::TokioClientConfig, processor::RequestProcessor, RPCHook,
-        ServiceBridge,
-    },
+    runtime::{config::client_config::TokioClientConfig, processor::RequestProcessor, RPCHook},
 };
 
 pub struct RocketmqDefaultClient {
-    service_bridge: ServiceBridge,
     tokio_client_config: TokioClientConfig,
     //cache connection
     connection_tables:
@@ -42,7 +39,6 @@ pub struct RocketmqDefaultClient {
 impl RocketmqDefaultClient {
     pub fn new(tokio_client_config: TokioClientConfig) -> Self {
         Self {
-            service_bridge: ServiceBridge::new(),
             tokio_client_config,
             connection_tables: Default::default(),
             namesrv_addr_list: Arc::new(Default::default()),
@@ -133,7 +129,8 @@ impl RemotingClient for RocketmqDefaultClient {
     }
 
     fn get_name_server_address_list(&self) -> Vec<String> {
-        todo!()
+        let cloned = self.namesrv_addr_list.clone();
+        Handle::current().block_on(async move { cloned.lock().await.clone() })
     }
 
     fn get_available_name_srv_list(&self) -> Vec<String> {
@@ -147,10 +144,15 @@ impl RemotingClient for RocketmqDefaultClient {
         timeout_millis: u64,
     ) -> RemotingCommand {
         let client = self.get_and_create_client(addr.clone()).await;
-        let client_ref = &mut *client.lock().await;
-        ServiceBridge::invoke_sync(client_ref, request, timeout_millis)
-            .await
-            .unwrap()
+        if let Ok(result) = timeout(Duration::from_millis(timeout_millis), async {
+            client.lock().await.send_read(request).await.unwrap()
+        })
+        .await
+        {
+            result
+        } else {
+            RemotingCommand::create_response_command()
+        }
     }
 
     async fn invoke_async(
@@ -161,18 +163,29 @@ impl RemotingClient for RocketmqDefaultClient {
         invoke_callback: impl InvokeCallback,
     ) -> Result<(), Box<dyn Error>> {
         let client = self.get_and_create_client(addr.clone()).await;
-        let client_ref = &mut *client.lock().await;
-        ServiceBridge::invoke_async(client_ref, request, timeout_millis, invoke_callback).await;
+        if let Ok(resp) = time::timeout(Duration::from_millis(timeout_millis), async {
+            client.lock().await.send_read(request).await.unwrap()
+        })
+        .await
+        {
+            invoke_callback.operation_succeed(resp)
+        }
+
         Ok(())
     }
 
-    fn invoke_oneway(
+    async fn invoke_oneway(
         &self,
         addr: String,
         request: RemotingCommand,
         timeout_millis: u64,
     ) -> Result<(), Box<dyn Error>> {
-        todo!()
+        let client = self.get_and_create_client(addr.clone()).await;
+        let _ = time::timeout(Duration::from_millis(timeout_millis), async move {
+            client.lock().await.send(request).await.unwrap()
+        })
+        .await;
+        Ok(())
     }
 
     fn register_processor(

rocketmq-remoting/src/runtime.rs

@@ -15,16 +15,11 @@
  * limitations under the License.
  */
 
-use std::{collections::HashMap, sync::Arc, time::Duration};
-
-use tokio::{time, time::timeout};
+use std::{collections::HashMap, sync::Arc};
 
 use crate::{
-    clients::Client,
-    net::ResponseFuture,
-    protocol::{remoting_command::RemotingCommand, RemotingCommandType},
-    remoting::InvokeCallback,
-    runtime::{processor::RequestProcessor, server::ConnectionHandlerContext},
+    net::ResponseFuture, protocol::remoting_command::RemotingCommand,
+    runtime::processor::RequestProcessor,
 };
 
 pub mod config;
@@ -80,7 +75,7 @@ impl Default for ServiceBridge {
 }
 
 impl ServiceBridge {
-    pub fn process_message_received(
+    /*    pub fn process_message_received(
         &mut self,
         ctx: ConnectionHandlerContext,
         msg: RemotingCommand,
@@ -103,25 +98,26 @@ impl ServiceBridge {
         _ctx: ConnectionHandlerContext,
         _msg: RemotingCommand,
     ) {
-    }
+    }*/
 
+    /*    #[allow(unused_variables)]
     pub async fn invoke_async(
         client: &mut Client,
         //client: Arc<Mutex<Client>>,
         request: RemotingCommand,
         timeout_millis: u64,
         invoke_callback: impl InvokeCallback,
     ) {
-        if let Ok(resp) = time::timeout(Duration::from_millis(timeout_millis), async {
+                if let Ok(resp) = time::timeout(Duration::from_millis(timeout_millis), async {
             client.invoke(request).await.unwrap()
         })
         .await
         {
             invoke_callback.operation_succeed(resp)
         }
-    }
+    }*/
 
-    pub async fn invoke_sync(
+    /*    pub async fn invoke_sync(
         client: &mut Client,
         request: RemotingCommand,
         timeout_millis: u64,
@@ -132,5 +128,5 @@ impl ServiceBridge {
         })
         .await;
         Some(result.unwrap())
-    }
+    }*/
 }