[ISSUE #75] 🚀 Namesrv support add write perm of brober (request code 327)

Author: mxsm

README.md

@@ -94,7 +94,7 @@ Feature list:
 | Get broker member_group                | 901          | :sparkling_heart: :white_check_mark: |        |
 | Get broker cluster info                | 106          | :sparkling_heart: :white_check_mark: |        |
 | Wipe write perm of boker               | 205          | :sparkling_heart: :white_check_mark: |        |
-| Add write perm of brober               | 327          | :broken_heart: :x: |        |
+| Add write perm of brober               | 327          | :sparkling_heart: :white_check_mark: |        |
 | Get all topic list from name server    | 206          | :broken_heart: :x: |        |
 | Delete topic in name server            | 216          | :broken_heart: :x: |        |
 | Register topic in name server          | 217          | :broken_heart: :x: |        |

rocketmq-namesrv/README.md

@@ -27,7 +27,7 @@ Feature list:
 | Get broker member_group                | 901          | :sparkling_heart: :white_check_mark: |        |
 | Get broker cluster info                | 106          | :sparkling_heart: :white_check_mark: |        |
 | Wipe write perm of boker               | 205          | :sparkling_heart: :white_check_mark: |        |
-| Add write perm of brober               | 327          | :broken_heart: :x: |        |
+| Add write perm of brober               | 327          | :sparkling_heart: :white_check_mark: |        |
 | Get all topic list from name server    | 206          | :broken_heart: :x: |        |
 | Delete topic in name server            | 216          | :broken_heart: :x: |        |
 | Register topic in name server          | 217          | :broken_heart: :x: |        |

rocketmq-namesrv/src/processor/default_request_processor.rs

@@ -42,6 +42,7 @@ use rocketmq_remoting::{
                 PutKVConfigRequestHeader,
             },
             perm_broker_header::{
+                AddWritePermOfBrokerRequestHeader, AddWritePermOfBrokerResponseHeader,
                 WipeWritePermOfBrokerRequestHeader, WipeWritePermOfBrokerResponseHeader,
             },
             query_data_version_header::{
@@ -81,6 +82,7 @@ impl RequestProcessor for DefaultRequestProcessor {
             //handle get broker cluster info
             Some(RequestCode::GetBrokerClusterInfo) => self.get_broker_cluster_info(request),
             Some(RequestCode::WipeWritePermOfBroker) => self.wipe_write_perm_of_broker(request),
+            Some(RequestCode::AddWritePermOfBroker) => self.add_write_perm_of_broker(request),
 
             _ => RemotingCommand::create_response_command_with_code(
                 RemotingSysResponseCode::SystemError,
@@ -333,6 +335,19 @@ impl DefaultRequestProcessor {
             WipeWritePermOfBrokerResponseHeader::new(wipe_topic_cnt),
         )))
     }
+
+    fn add_write_perm_of_broker(&mut self, request: RemotingCommand) -> RemotingCommand {
+        let request_header = request
+            .decode_command_custom_header::<AddWritePermOfBrokerRequestHeader>()
+            .unwrap();
+        let add_topic_cnt = self
+            .route_info_manager
+            .write()
+            .add_write_perm_of_broker_by_lock(request_header.broker_name.as_str());
+        RemotingCommand::create_response_command().set_command_custom_header(Some(Box::new(
+            AddWritePermOfBrokerResponseHeader::new(add_topic_cnt),
+        )))
+    }
 }
 
 fn extract_register_topic_config_from_request(

rocketmq-namesrv/src/route/route_info_manager.rs

@@ -648,6 +648,10 @@ impl RouteInfoManager {
         self.operate_write_perm_of_broker(broker_name, RequestCode::WipeWritePermOfBroker)
     }
 
+    pub(crate) fn add_write_perm_of_broker_by_lock(&mut self, broker_name: &str) -> i32 {
+        self.operate_write_perm_of_broker(broker_name, RequestCode::AddWritePermOfBroker)
+    }
+
     fn operate_write_perm_of_broker(
         &mut self,
         broker_name: &str,

rocketmq-remoting/src/protocol/header/namesrv/perm_broker_header.rs

@@ -92,3 +92,76 @@ impl FromMap for WipeWritePermOfBrokerResponseHeader {
         })
     }
 }
+
+#[derive(Debug, Clone, Deserialize, Serialize, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct AddWritePermOfBrokerRequestHeader {
+    pub broker_name: String,
+}
+
+impl AddWritePermOfBrokerRequestHeader {
+    const BROKER_NAME: &'static str = "brokerName";
+    pub fn new(broker_name: impl Into<String>) -> Self {
+        Self {
+            broker_name: broker_name.into(),
+        }
+    }
+}
+
+impl CommandCustomHeader for AddWritePermOfBrokerRequestHeader {
+    fn to_map(&self) -> Option<HashMap<String, String>> {
+        Some(HashMap::from([(
+            Self::BROKER_NAME.to_string(),
+            self.broker_name.clone(),
+        )]))
+    }
+}
+
+impl FromMap for AddWritePermOfBrokerRequestHeader {
+    type Target = Self;
+
+    fn from(map: &HashMap<String, String>) -> Option<Self::Target> {
+        Some(AddWritePermOfBrokerRequestHeader {
+            broker_name: map
+                .get(AddWritePermOfBrokerRequestHeader::BROKER_NAME)
+                .cloned()
+                .unwrap_or_default(),
+        })
+    }
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct AddWritePermOfBrokerResponseHeader {
+    pub add_topic_count: i32,
+}
+
+impl AddWritePermOfBrokerResponseHeader {
+    const ADD_TOPIC_COUNT: &'static str = "addTopicCount";
+
+    pub fn new(add_topic_count: i32) -> Self {
+        Self { add_topic_count }
+    }
+}
+
+impl CommandCustomHeader for AddWritePermOfBrokerResponseHeader {
+    fn to_map(&self) -> Option<HashMap<String, String>> {
+        Some(HashMap::from([(
+            Self::ADD_TOPIC_COUNT.to_string(),
+            self.add_topic_count.to_string(),
+        )]))
+    }
+}
+
+impl FromMap for AddWritePermOfBrokerResponseHeader {
+    type Target = Self;
+
+    fn from(map: &HashMap<String, String>) -> Option<Self::Target> {
+        Some(AddWritePermOfBrokerResponseHeader {
+            add_topic_count: map
+                .get(AddWritePermOfBrokerResponseHeader::ADD_TOPIC_COUNT)
+                .and_then(|s| s.parse::<i32>().ok())
+                .unwrap_or(0),
+        })
+    }
+}