[ISSUE #191]🚀Implementing network communication for the Broker-3 (#192)

Author: mxsm

rocketmq-broker/src/bin/broker_bootstrap_server.rs

@@ -18,10 +18,10 @@
 use std::{path::PathBuf, process::exit};
 
 use clap::Parser;
-use rocketmq_broker::{broker_controller::BrokerController, command::Args};
-use rocketmq_common::{
-    common::broker::broker_config::BrokerConfig, EnvUtils::EnvUtils, ParseConfigFile,
+use rocketmq_broker::{
+    broker_config::BrokerConfig, broker_controller::BrokerController, command::Args,
 };
+use rocketmq_common::{EnvUtils::EnvUtils, ParseConfigFile};
 use rocketmq_rust::rocketmq;
 use rocketmq_store::config::message_store_config::MessageStoreConfig;
 use tracing::info;

rocketmq-broker/src/broker_config.rs

@@ -0,0 +1,67 @@
+/*
+ * 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_common::common::{
+    broker::broker_config::{BrokerIdentity, TimerWheelConfig, TopicConfig, TopicQueueConfig},
+    topic::TopicValidator,
+};
+use rocketmq_remoting::server::config::BrokerServerConfig;
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BrokerConfig {
+    pub broker_identity: BrokerIdentity,
+
+    pub topic_config: TopicConfig,
+
+    pub topic_queue_config: TopicQueueConfig,
+
+    pub timer_wheel_config: TimerWheelConfig,
+
+    pub broker_server_config: BrokerServerConfig,
+
+    pub broker_ip1: String,
+    pub broker_ip2: Option<String>,
+    pub listen_port: u32,
+    pub trace_topic_enable: bool,
+    pub msg_trace_topic_name: String,
+    pub enable_controller_mode: bool,
+}
+
+impl Default for BrokerConfig {
+    fn default() -> Self {
+        let broker_identity = BrokerIdentity::new();
+        let broker_ip1 = String::from("127.0.0.1");
+        let broker_ip2 = None;
+        let listen_port = 10911;
+
+        BrokerConfig {
+            broker_identity,
+            topic_config: TopicConfig::default(),
+            topic_queue_config: TopicQueueConfig::default(),
+            timer_wheel_config: TimerWheelConfig::default(),
+            broker_server_config: Default::default(),
+            broker_ip1,
+            broker_ip2,
+            listen_port,
+            trace_topic_enable: false,
+            msg_trace_topic_name: TopicValidator::RMQ_SYS_TRACE_TOPIC.to_string(),
+            enable_controller_mode: false,
+        }
+    }
+}

rocketmq-broker/src/broker_controller.rs

@@ -16,7 +16,10 @@
  */
 use std::net::SocketAddr;
 
-use rocketmq_common::common::{broker::broker_config::BrokerConfig, config_manager::ConfigManager};
+use rocketmq_common::common::config_manager::ConfigManager;
+use rocketmq_remoting::{
+    remoting::RemotingService, server::rocketmq_server::RocketmqDefaultServer,
+};
 use rocketmq_store::{
     base::store_enum::StoreType, config::message_store_config::MessageStoreConfig,
     log_file::MessageStore, message_store::local_file_store::LocalFileMessageStore,
@@ -26,6 +29,7 @@ use rocketmq_store::{
 use tracing::{info, warn};
 
 use crate::{
+    broker_config::BrokerConfig,
     broker_outer_api::BrokerOuterAPI,
     client::{
         default_consumer_ids_change_listener::DefaultConsumerIdsChangeListener,
@@ -36,6 +40,7 @@ use crate::{
         cold_data_cg_ctr_service::ColdDataCgCtrService,
         cold_data_pull_request_hold_service::ColdDataPullRequestHoldService,
     },
+    controller::replicas_manager::ReplicasManager,
     filter::manager::consumer_filter_manager::ConsumerFilterManager,
     longpolling::{
         longpolling_service::pull_request_hold_service::PullRequestHoldService,
@@ -96,6 +101,9 @@ pub struct BrokerController {
     pub(crate) broker_outer_api: BrokerOuterAPI,
     pub(crate) message_store: Option<Box<dyn MessageStore>>,
     pub(crate) timer_message_store: Option<TimerMessageStore>,
+    pub(crate) replicas_manager: Option<ReplicasManager>,
+    pub(crate) broker_server: Option<RocketmqDefaultServer>,
+    pub(crate) fast_broker_server: Option<RocketmqDefaultServer>,
 }
 
 impl BrokerController {
@@ -137,12 +145,33 @@ impl BrokerController {
             broker_outer_api: Default::default(),
             message_store: None,
             timer_message_store: None,
+            replicas_manager: None,
+            broker_server: None,
+            fast_broker_server: None,
         }
     }
 }
 
 impl BrokerController {
-    pub async fn start(&mut self) {}
+    pub async fn start(&mut self) {
+        if self.message_store.is_some() {
+            let _ = self.message_store.as_mut().unwrap().start();
+        }
+
+        if let Some(ref mut timer_message_store) = self.timer_message_store {
+            timer_message_store.start();
+        }
+
+        if let Some(ref mut replicas_manager) = self.replicas_manager {
+            replicas_manager.start();
+        }
+
+        if let Some(ref mut broker_server) = self.broker_server {
+            broker_server.start().await;
+        }
+
+        //other service start
+    }
 
     pub fn initialize(&mut self) -> bool {
         let mut result = self.initialize_metadata();
@@ -206,7 +235,13 @@ impl BrokerController {
         result
     }
 
-    fn initialize_remoting_server(&mut self) {}
+    fn initialize_remoting_server(&mut self) {
+        let broker_server =
+            RocketmqDefaultServer::new(self.broker_config.broker_server_config.clone());
+        self.broker_server = Some(broker_server);
+
+        // fast broker server implementation in future versions
+    }
 
     fn initialize_resources(&mut self) {}
 

rocketmq-broker/src/controller.rs

@@ -0,0 +1,17 @@
+/*
+ * 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.
+ */
+pub(crate) mod replicas_manager;

rocketmq-broker/src/controller/replicas_manager.rs

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+#[derive(Default)]
+pub struct ReplicasManager {}
+
+impl ReplicasManager {
+    pub fn start(&mut self) {}
+}

rocketmq-broker/src/lib.rs

@@ -19,9 +19,11 @@
 pub mod broker_controller;
 pub mod command;
 
+pub mod broker_config;
 mod broker_outer_api;
 mod client;
 mod coldctr;
+mod controller;
 mod filter;
 mod longpolling;
 mod offset;

rocketmq-broker/src/topic/manager/topic_config_manager.rs

@@ -18,11 +18,13 @@
 use std::{collections::HashMap, sync::Arc};
 
 use rocketmq_common::common::{
-    broker::broker_config::BrokerConfig, config::TopicConfig, config_manager::ConfigManager,
-    constant::PermName, mix_all, topic::TopicValidator,
+    config::TopicConfig, config_manager::ConfigManager, constant::PermName, mix_all,
+    topic::TopicValidator,
 };
 use rocketmq_remoting::protocol::DataVersion;
 
+use crate::broker_config::BrokerConfig;
+
 #[derive(Default)]
 pub(crate) struct TopicConfigManager {
     pub consumer_order_info_manager: String,

rocketmq-common/src/common/broker/broker_config.rs

@@ -30,7 +30,7 @@ pub struct BrokerIdentity {
 }
 
 impl BrokerIdentity {
-    fn new() -> Self {
+    pub fn new() -> Self {
         let broker_name = default_broker_name();
         let broker_cluster_name = String::from("DefaultCluster");
         let broker_id = mix_all::MASTER_ID;
@@ -86,47 +86,6 @@ fn default_broker_name() -> String {
     String::from("DefaultBrokerName")
 }
 
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct BrokerConfig {
-    pub broker_identity: BrokerIdentity,
-
-    pub topic_config: TopicConfig,
-
-    pub topic_queue_config: TopicQueueConfig,
-
-    pub timer_wheel_config: TimerWheelConfig,
-
-    pub broker_ip1: String,
-    pub broker_ip2: Option<String>,
-    pub listen_port: u32,
-    pub trace_topic_enable: bool,
-    pub msg_trace_topic_name: String,
-    pub enable_controller_mode: bool,
-}
-
-impl Default for BrokerConfig {
-    fn default() -> Self {
-        let broker_identity = BrokerIdentity::new();
-        let broker_ip1 = String::from("127.0.0.1");
-        let broker_ip2 = None;
-        let listen_port = 10911;
-
-        BrokerConfig {
-            broker_identity,
-            topic_config: TopicConfig::default(),
-            topic_queue_config: TopicQueueConfig::default(),
-            timer_wheel_config: TimerWheelConfig::default(),
-            broker_ip1,
-            broker_ip2,
-            listen_port,
-            trace_topic_enable: false,
-            msg_trace_topic_name: TopicValidator::RMQ_SYS_TRACE_TOPIC.to_string(),
-            enable_controller_mode: false,
-        }
-    }
-}
-
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct TopicConfig {

rocketmq-remoting/src/runtime.rs

@@ -31,3 +31,5 @@ pub trait RPCHook {
         response: &RemotingCommand,
     );
 }
+
+pub struct ServerInner {}

rocketmq-remoting/src/server.rs

@@ -15,6 +15,7 @@
  * limitations under the License.
  */
 
+pub mod config;
 pub mod rocketmq_server;
 
 use std::sync::Arc;

rocketmq-remoting/src/server/config.rs

@@ -0,0 +1,21 @@
+/*
+ * 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 serde::{Deserialize, Serialize};
+
+#[derive(Default, Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BrokerServerConfig {}