[ISSUE #169]🚀Implementing network communication for the Broker (#205)

* [ISSUE #169]🚀Implementing network communication for the Broker

* add test case

Author: mxsm

rocketmq-broker/src/bin/broker_bootstrap_server.rs

@@ -67,7 +67,6 @@ async fn start_broker_controller(broker_controller: BrokerController) -> anyhow:
         "Rocketmq name server(Rust) running on {}:{}",
         broker_controller.broker_config.broker_ip1, broker_controller.broker_config.listen_port,
     );
-    let future = broker_controller.start();
-    join!(future);
+    join!(broker_controller.start());
     Ok(())
 }

rocketmq-broker/src/broker_controller.rs

@@ -174,7 +174,7 @@ impl BrokerController {
 
         if let Some(ref mut broker_server) = self.broker_server {
             broker_server.start().await;
-        };
+        }
 
         //other service start
     }

rocketmq-broker/src/lib.rs

@@ -26,6 +26,7 @@ mod coldctr;
 mod controller;
 mod filter;
 mod longpolling;
+mod mqtrace;
 mod offset;
 mod processor;
 mod schedule;

rocketmq-broker/src/mqtrace.rs

@@ -0,0 +1,18 @@
+/*
+ * 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 send_message_context;

rocketmq-broker/src/mqtrace/send_message_context.rs

@@ -0,0 +1,56 @@
+/*
+ * 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 std::{any::Any, collections::HashMap};
+
+use rocketmq_common::common::message::message_enum::MessageType;
+use rocketmq_store::status::StatsType;
+
+#[derive(Debug, Default)]
+pub struct SendMessageContext {
+    pub namespace: String,
+    pub producer_group: String,
+    pub topic: String,
+    pub msg_id: String,
+    pub origin_msg_id: String,
+    pub queue_id: Option<i32>,
+    pub queue_offset: Option<i64>,
+    pub broker_addr: String,
+    pub born_host: String,
+    pub body_length: i32,
+    pub code: i32,
+    pub error_msg: String,
+    pub msg_props: String,
+    pub mq_trace_context: Option<Box<dyn Any>>,
+    pub ext_props: HashMap<String, String>,
+    pub broker_region_id: String,
+    pub msg_unique_key: String,
+    pub born_time_stamp: i64,
+    pub request_time_stamp: i64,
+    pub msg_type: MessageType,
+    pub is_success: bool,
+    pub account_auth_type: String,
+    pub account_owner_parent: String,
+    pub account_owner_self: String,
+    pub send_msg_num: i32,
+    pub send_msg_size: i32,
+    pub send_stat: StatsType,
+    pub commercial_send_msg_num: i32,
+    pub commercial_owner: String,
+    pub commercial_send_stats: StatsType,
+    pub commercial_send_size: i32,
+    pub commercial_send_times: i32,
+}

rocketmq-broker/src/processor/send_message_processor.rs

@@ -15,19 +15,45 @@
  * limitations under the License.
  */
 use rocketmq_remoting::{
-    protocol::remoting_command::RemotingCommand,
+    code::request_code::RequestCode,
+    protocol::{
+        header::message_operation_header::send_message_request_header::parse_request_header,
+        remoting_command::RemotingCommand,
+    },
     runtime::{processor::RequestProcessor, server::ConnectionHandlerContext},
 };
 
 #[derive(Default)]
-pub struct SendMessageProcessor {}
+pub struct SendMessageProcessor {
+    inner: SendMessageProcessorInner,
+}
 
 impl RequestProcessor for SendMessageProcessor {
     fn process_request(
         &mut self,
-        _ctx: ConnectionHandlerContext,
-        _request: RemotingCommand,
+        ctx: ConnectionHandlerContext,
+        request: RemotingCommand,
     ) -> RemotingCommand {
+        let request_code = RequestCode::from(request.code());
+        match request_code {
+            RequestCode::ConsumerSendMsgBack => self.inner.consumer_send_msg_back(&ctx, &request),
+            _ => {
+                let _request_header = parse_request_header(&request).unwrap();
+            }
+        }
+        RemotingCommand::create_response_command()
+    }
+}
+
+#[derive(Default)]
+struct SendMessageProcessorInner {}
+
+impl SendMessageProcessorInner {
+    fn consumer_send_msg_back(
+        &mut self,
+        _ctx: &ConnectionHandlerContext,
+        _request: &RemotingCommand,
+    ) {
         todo!()
     }
 }

rocketmq-common/src/common/message.rs

@@ -23,6 +23,7 @@ use std::{
 use lazy_static::lazy_static;
 
 pub mod message_batch;
+pub mod message_enum;
 pub mod message_id;
 pub mod message_queue;
 pub mod message_single;

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

@@ -0,0 +1,111 @@
+/*
+ * 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(Debug, PartialEq, Copy, Clone, Default)]
+pub enum MessageType {
+    #[default]
+    NormalMsg,
+    TransMsgHalf,
+    TransMsgCommit,
+    DelayMsg,
+    OrderMsg,
+}
+
+impl MessageType {
+    pub fn get_short_name(&self) -> &'static str {
+        match self {
+            MessageType::NormalMsg => "Normal",
+            MessageType::TransMsgHalf => "Trans",
+            MessageType::TransMsgCommit => "TransCommit",
+            MessageType::DelayMsg => "Delay",
+            MessageType::OrderMsg => "Order",
+        }
+    }
+
+    pub fn get_by_short_name(short_name: &str) -> MessageType {
+        match short_name {
+            "Normal" => MessageType::NormalMsg,
+            "Trans" => MessageType::TransMsgHalf,
+            "TransCommit" => MessageType::TransMsgCommit,
+            "Delay" => MessageType::DelayMsg,
+            "Order" => MessageType::OrderMsg,
+            _ => MessageType::NormalMsg,
+        }
+    }
+}
+
+#[derive(Debug, PartialEq, Copy, Clone)]
+pub enum MessageRequestMode {
+    Pull,
+    Pop,
+}
+
+impl MessageRequestMode {
+    pub fn get_name(&self) -> &'static str {
+        match self {
+            MessageRequestMode::Pull => "PULL",
+            MessageRequestMode::Pop => "POP",
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_get_short_name() {
+        assert_eq!(MessageType::NormalMsg.get_short_name(), "Normal");
+        assert_eq!(MessageType::TransMsgHalf.get_short_name(), "Trans");
+        assert_eq!(MessageType::TransMsgCommit.get_short_name(), "TransCommit");
+        assert_eq!(MessageType::DelayMsg.get_short_name(), "Delay");
+        assert_eq!(MessageType::OrderMsg.get_short_name(), "Order");
+    }
+
+    #[test]
+    fn test_get_by_short_name() {
+        assert_eq!(
+            MessageType::get_by_short_name("Normal"),
+            MessageType::NormalMsg
+        );
+        assert_eq!(
+            MessageType::get_by_short_name("Trans"),
+            MessageType::TransMsgHalf
+        );
+        assert_eq!(
+            MessageType::get_by_short_name("TransCommit"),
+            MessageType::TransMsgCommit
+        );
+        assert_eq!(
+            MessageType::get_by_short_name("Delay"),
+            MessageType::DelayMsg
+        );
+        assert_eq!(
+            MessageType::get_by_short_name("Order"),
+            MessageType::OrderMsg
+        );
+        assert_eq!(
+            MessageType::get_by_short_name("Invalid"),
+            MessageType::NormalMsg
+        );
+    }
+
+    #[test]
+    fn test_get_name() {
+        assert_eq!(MessageRequestMode::Pull.get_name(), "PULL");
+        assert_eq!(MessageRequestMode::Pop.get_name(), "POP");
+    }
+}

rocketmq-remoting/src/lib.rs

@@ -1,3 +1,4 @@
+#![feature(future_join)]
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
@@ -23,6 +24,7 @@ pub mod error;
 pub mod net;
 pub mod protocol;
 pub mod remoting;
+pub mod rpc;
 pub mod runtime;
 pub mod server;
 

rocketmq-remoting/src/protocol/header.rs

@@ -16,4 +16,5 @@
  */
 pub mod broker;
 pub mod client_request_header;
+pub mod message_operation_header;
 pub mod namesrv;

rocketmq-remoting/src/protocol/header/message_operation_header.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 mod send_message_request_header;