[ISSUE #295]Refactor network section code (#296)

Author: mxsm

Cargo.toml

@@ -6,7 +6,7 @@ members = [
     "rocketmq-filter",
     "rocketmq-macros",
     "rocketmq-namesrv",
-    "rocketmq-remoting",
+    "rocketmq-remoting", "rocketmq-runtime",
     "rocketmq-store"]
 resolver = "2"
 
@@ -51,4 +51,4 @@ num_cpus = "1.16"
 config = "0.14"
 
 parking_lot = "0.12"
-dirs = "5.0"
+dirs = "5.0"

rocketmq-broker/Cargo.toml

@@ -23,6 +23,7 @@ rocketmq-remoting = { version = "0.2.0", path = "../rocketmq-remoting" }
 rocketmq-store = { version = "0.2.0", path = "../rocketmq-store", default-features = true }
 rocketmq-filter = { version = "0.2.0", path = "../rocketmq-filter" }
 rocketmq-macros = { version = "0.2.0", path = "../rocketmq-macros" }
+rocketmq-runtime = { version = "0.2.0", path = "../rocketmq-runtime" }
 
 anyhow.workspace = true
 env_logger.workspace = true

rocketmq-broker/src/bin/broker_bootstrap_server.rs

@@ -15,59 +15,56 @@
  * limitations under the License.
  */
 
-use std::{path::PathBuf, process::exit};
+use std::path::PathBuf;
 
 use clap::Parser;
-use rocketmq_broker::{
-    broker_config::BrokerConfig, broker_controller::BrokerController, command::Args,
-};
+use rocketmq_broker::{broker_config::BrokerConfig, command::Args, Builder};
 use rocketmq_common::{EnvUtils::EnvUtils, ParseConfigFile};
 use rocketmq_rust::rocketmq;
 use rocketmq_store::config::message_store_config::MessageStoreConfig;
-use tracing::{info, warn};
+use tracing::info;
 
 #[rocketmq::main]
 async fn main() -> anyhow::Result<()> {
     // init logger
     rocketmq_common::log::init_logger();
-    let controller = create_broker_controller()?;
-    start_broker_controller(controller).await?;
+    let (broker_config, message_store_config) = parse_config_file();
+    // boot strap broker
+    Builder::new()
+        .set_broker_config(broker_config)
+        .set_message_store_config(message_store_config)
+        .build()
+        .boot()
+        .await;
     Ok(())
 }
 
-fn create_broker_controller() -> anyhow::Result<BrokerController> {
+fn parse_config_file() -> (BrokerConfig, MessageStoreConfig) {
     let args = Args::parse();
     let home = EnvUtils::get_rocketmq_home();
-    let (broker_config, message_store_config) = if let Some(ref config_file) = args.config_file {
+    let config = if let Some(ref config_file) = args.config_file {
         let config_file = PathBuf::from(config_file);
         (
-            ParseConfigFile::parse_config_file::<BrokerConfig>(config_file.clone())?,
-            ParseConfigFile::parse_config_file::<MessageStoreConfig>(config_file.clone())?,
+            ParseConfigFile::parse_config_file::<BrokerConfig>(config_file.clone())
+                .ok()
+                .unwrap(),
+            ParseConfigFile::parse_config_file::<MessageStoreConfig>(config_file.clone())
+                .ok()
+                .unwrap(),
         )
     } else {
         let path_buf = PathBuf::from(home.as_str())
             .join("conf")
             .join("broker.toml");
         (
-            ParseConfigFile::parse_config_file::<BrokerConfig>(path_buf.clone())?,
-            ParseConfigFile::parse_config_file::<MessageStoreConfig>(path_buf)?,
+            ParseConfigFile::parse_config_file::<BrokerConfig>(path_buf.clone())
+                .ok()
+                .unwrap(),
+            ParseConfigFile::parse_config_file::<MessageStoreConfig>(path_buf)
+                .ok()
+                .unwrap(),
         )
     };
     info!("Rocketmq(Rust) home: {}", home);
-    Ok(BrokerController::new(broker_config, message_store_config))
-}
-
-async fn start_broker_controller(broker_controller: BrokerController) -> anyhow::Result<()> {
-    let mut broker_controller = broker_controller;
-    if !broker_controller.initialize() {
-        warn!("Rocketmq(Rust) start failed, initialize failed");
-        exit(0);
-    }
-    info!(
-        "Rocketmq name server(Rust) running on {}:{}",
-        broker_controller.broker_config.broker_ip1, broker_controller.broker_config.listen_port,
-    );
-    //start controller
-    broker_controller.start().await;
-    Ok(())
+    config
 }

rocketmq-broker/src/broker_bootstrap.rs

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 rocketmq_remoting::server::config::ServerConfig;
+use rocketmq_store::config::message_store_config::MessageStoreConfig;
+use tokio::select;
+use tracing::error;
+
+use crate::{broker_config::BrokerConfig, broker_runtime::BrokerRuntime};
+
+pub struct BrokerBootstrap {
+    broker_runtime: BrokerRuntime,
+}
+
+impl BrokerBootstrap {
+    pub async fn boot(mut self) {
+        if !self.initialize().await {
+            error!("initialize fail");
+            return;
+        }
+
+        select! {
+            _ = self.start() =>{
+
+            }
+        }
+    }
+
+    async fn initialize(&mut self) -> bool {
+        self.broker_runtime.initialize().await
+    }
+
+    async fn start(&mut self) {
+        self.broker_runtime.start().await;
+    }
+}
+
+pub struct Builder {
+    broker_config: BrokerConfig,
+    message_store_config: MessageStoreConfig,
+    server_config: ServerConfig,
+}
+
+impl Builder {
+    pub fn new() -> Self {
+        Builder {
+            broker_config: Default::default(),
+            message_store_config: MessageStoreConfig::new(),
+            server_config: Default::default(),
+        }
+    }
+
+    pub fn set_broker_config(mut self, broker_config: BrokerConfig) -> Self {
+        self.broker_config = broker_config;
+        self
+    }
+    pub fn set_message_store_config(mut self, message_store_config: MessageStoreConfig) -> Self {
+        self.message_store_config = message_store_config;
+        self
+    }
+
+    pub fn set_server_config(mut self, server_config: ServerConfig) -> Self {
+        self.server_config = server_config;
+        self
+    }
+
+    pub fn build(self) -> BrokerBootstrap {
+        BrokerBootstrap {
+            broker_runtime: BrokerRuntime::new(
+                self.broker_config,
+                self.message_store_config,
+                self.server_config,
+            ),
+        }
+    }
+}
+
+impl Default for Builder {
+    fn default() -> Self {
+        Self::new()
+    }
+}