[ISSUE #82]Namesrv support delete topic in namesrv (request code 216)

Author: mxsm

rocketmq-namesrv/src/processor/default_request_processor.rs

@@ -49,6 +49,7 @@ use rocketmq_remoting::{
                 QueryDataVersionRequestHeader, QueryDataVersionResponseHeader,
             },
             register_broker_header::{RegisterBrokerRequestHeader, RegisterBrokerResponseHeader},
+            topic_operation_header::DeleteTopicFromNamesrvRequestHeader,
         },
         remoting_command::RemotingCommand,
         DataVersion, RemotingSerializable,
@@ -86,7 +87,7 @@ impl RequestProcessor for DefaultRequestProcessor {
             Some(RequestCode::GetAllTopicListFromNameserver) => {
                 self.get_all_topic_list_from_nameserver(request)
             }
-
+            Some(RequestCode::DeleteTopicInNamesrv) => self.delete_topic_in_name_srv(request),
             _ => RemotingCommand::create_response_command_with_code(
                 RemotingSysResponseCode::SystemError,
             ),
@@ -363,6 +364,17 @@ impl DefaultRequestProcessor {
         RemotingCommand::create_response_command_with_code(RemotingSysResponseCode::SystemError)
             .set_remark(Some(String::from("disable")))
     }
+
+    fn delete_topic_in_name_srv(&mut self, request: RemotingCommand) -> RemotingCommand {
+        let request_header = request
+            .decode_command_custom_header::<DeleteTopicFromNamesrvRequestHeader>()
+            .unwrap();
+        self.route_info_manager.write().delete_topic(
+            request_header.topic.as_str(),
+            request_header.cluster_name.clone(),
+        );
+        RemotingCommand::create_response_command()
+    }
 }
 
 fn extract_register_topic_config_from_request(

rocketmq-namesrv/src/route/route_info_manager.rs

@@ -689,4 +689,31 @@ impl RouteInfoManager {
             broker_addr: None,
         }
     }
+
+    pub(crate) fn delete_topic(
+        &mut self,
+        topic: impl Into<String>,
+        cluster_name: Option<impl Into<String>>,
+    ) {
+        let topic_inner = topic.into();
+        if cluster_name.is_some() {
+            let cluster_name_inner = cluster_name.map(|s| s.into()).unwrap();
+            let broker_names = self.cluster_addr_table.get(cluster_name_inner.as_str());
+            if broker_names.is_none() || broker_names.unwrap().is_empty() {
+                return;
+            }
+            if let Some(queue_data_map) = self.topic_queue_table.get_mut(topic_inner.as_str()) {
+                for broker_name in broker_names.unwrap() {
+                    if let Some(remove_qd) = queue_data_map.remove(broker_name) {
+                        info!(
+                            "deleteTopic, remove one broker's topic {} {} {:?}",
+                            broker_name, &topic_inner, remove_qd
+                        )
+                    }
+                }
+            }
+        } else {
+            self.topic_queue_table.remove(topic_inner.as_str());
+        }
+    }
 }

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

@@ -21,3 +21,4 @@ pub mod kv_config_header;
 pub mod perm_broker_header;
 pub mod query_data_version_header;
 pub mod register_broker_header;
+pub mod topic_operation_header;

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

@@ -0,0 +1,61 @@
+/*
+ * 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::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::protocol::command_custom_header::{CommandCustomHeader, FromMap};
+
+#[derive(Debug, Clone, Deserialize, Serialize, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct DeleteTopicFromNamesrvRequestHeader {
+    pub topic: String,
+    pub cluster_name: Option<String>,
+}
+
+impl DeleteTopicFromNamesrvRequestHeader {
+    const TOPIC: &'static str = "topic";
+    const CLUSTER_NAME: &'static str = "clusterName";
+    pub fn new(topic: impl Into<String>, cluster_name: Option<impl Into<String>>) -> Self {
+        Self {
+            topic: topic.into(),
+            cluster_name: cluster_name.map(|s| s.into()),
+        }
+    }
+}
+
+impl CommandCustomHeader for DeleteTopicFromNamesrvRequestHeader {
+    fn to_map(&self) -> Option<HashMap<String, String>> {
+        let mut map = HashMap::from([(Self::TOPIC.to_string(), self.topic.clone())]);
+        if let Some(ref cluster_name) = self.cluster_name {
+            map.insert(Self::CLUSTER_NAME.to_string(), cluster_name.clone());
+        }
+        Some(map)
+    }
+}
+
+impl FromMap for DeleteTopicFromNamesrvRequestHeader {
+    type Target = Self;
+
+    fn from(map: &HashMap<String, String>) -> Option<Self::Target> {
+        Some(DeleteTopicFromNamesrvRequestHeader {
+            topic: map.get(Self::TOPIC).cloned().unwrap_or_default(),
+            cluster_name: map.get(Self::CLUSTER_NAME).map(|s| s.into()),
+        })
+    }
+}