[ISSUE #84] ⚡ Namesrv support get kv list by namespace (request code 219) (#94)

Author: mxsm

README.md

@@ -86,7 +86,7 @@ Feature list:
 | Put KV Config                          | 100          | :sparkling_heart: :white_check_mark: |        |
 | Get KV Config                          | 101          | :sparkling_heart: :white_check_mark: |        |
 | Delete KV Config                       | 102          | :sparkling_heart: :white_check_mark: |        |
-| Get kv list by namespace               | 219          | :broken_heart: :x: |        |
+| Get kv list by namespace               | 219          | :sparkling_heart: :white_check_mark: |        |
 | Query Data Version                     | 322          | :sparkling_heart: :white_check_mark: |        |
 | Register Broker                        | 103          | :heart:        |        |
 | Unregister Broker                      | 104          | :broken_heart: :x: |        |
@@ -97,7 +97,7 @@ Feature list:
 | Add write perm of brober               | 327          | :sparkling_heart: :white_check_mark: |        |
 | Get all topic list from name server    | 206          | :sparkling_heart: :white_check_mark: |        |
 | Delete topic in name server            | 216          | :sparkling_heart: :white_check_mark: |        |
-| Register topic in name server          | 217          | :broken_heart: :x: |        |
+| Register topic in name server          | 217          | :sparkling_heart: :white_check_mark: |        |
 | Get topics by cluster                  | 224          | :broken_heart: :x: |        |
 | Get system topic list from name server | 304          | :broken_heart: :x: |        |
 | Get unit topic list                    | 311          | :broken_heart: :x: |        |

rocketmq-namesrv/README.md

@@ -19,7 +19,7 @@ Feature list:
 | Put KV Config                          | 100          | :sparkling_heart: :white_check_mark: |        |
 | Get KV Config                          | 101          | :sparkling_heart: :white_check_mark: |        |
 | Delete KV Config                       | 102          | :sparkling_heart: :white_check_mark: |        |
-| Get kv list by namespace               | 219          | :broken_heart: :x:                   |        |
+| Get kv list by namespace               | 219          | :sparkling_heart: :white_check_mark: |        |
 | Query Data Version                     | 322          | :sparkling_heart: :white_check_mark: |        |
 | Register Broker                        | 103          | :heart: :white_check_mark:           |        |
 | Unregister Broker                      | 104          | :broken_heart: :x:                   |        |
@@ -30,7 +30,7 @@ Feature list:
 | Add write perm of brober               | 327          | :sparkling_heart: :white_check_mark: |        |
 | Get all topic list from name server    | 206          | :sparkling_heart: :white_check_mark: |        |
 | Delete topic in name server            | 216          | :sparkling_heart: :white_check_mark: |        |
-| Register topic in name server          | 217          | :broken_heart: :x:                   |        |
+| Register topic in name server          | 217          | :sparkling_heart: :white_check_mark: |        |
 | Get topics by cluster                  | 224          | :broken_heart: :x:                   |        |
 | Get system topic list from name server | 304          | :broken_heart: :x:                   |        |
 | Get unit topic list                    | 311          | :broken_heart: :x:                   |        |

rocketmq-namesrv/src/processor/default_request_processor.rs

@@ -23,7 +23,10 @@ use rocketmq_common::{
     CRC32Utils,
 };
 use rocketmq_remoting::{
-    code::{request_code::RequestCode, response_code::RemotingSysResponseCode},
+    code::{
+        request_code::RequestCode,
+        response_code::{RemotingSysResponseCode, ResponseCode},
+    },
     protocol::{
         body::{
             broker_body::{
@@ -39,7 +42,7 @@ use rocketmq_remoting::{
             },
             kv_config_header::{
                 DeleteKVConfigRequestHeader, GetKVConfigRequestHeader, GetKVConfigResponseHeader,
-                PutKVConfigRequestHeader,
+                GetKVListByNamespaceRequestHeader, PutKVConfigRequestHeader,
             },
             perm_broker_header::{
                 AddWritePermOfBrokerRequestHeader, AddWritePermOfBrokerResponseHeader,
@@ -92,6 +95,7 @@ impl RequestProcessor for DefaultRequestProcessor {
             }
             Some(RequestCode::DeleteTopicInNamesrv) => self.delete_topic_in_name_srv(request),
             Some(RequestCode::RegisterTopicInNamesrv) => self.register_topic_to_name_srv(request),
+            Some(RequestCode::GetKvlistByNamespace) => self.get_kv_list_by_namespace(request),
             _ => RemotingCommand::create_response_command_with_code(
                 RemotingSysResponseCode::SystemError,
             ),
@@ -394,6 +398,25 @@ impl DefaultRequestProcessor {
         }
         RemotingCommand::create_response_command()
     }
+
+    fn get_kv_list_by_namespace(&mut self, request: RemotingCommand) -> RemotingCommand {
+        let request_header = request
+            .decode_command_custom_header::<GetKVListByNamespaceRequestHeader>()
+            .unwrap();
+        let value = self
+            .kvconfig_manager
+            .write()
+            .get_kv_list_by_namespace(request_header.namespace.as_str());
+        if value.is_some() {
+            return RemotingCommand::create_response_command().set_body(value);
+        }
+        RemotingCommand::create_response_command_with_code(ResponseCode::QueryNotFound).set_remark(
+            Some(format!(
+                "No config item, Namespace: {}",
+                request_header.namespace.as_str()
+            )),
+        )
+    }
 }
 
 fn extract_register_topic_config_from_request(

rocketmq-remoting/src/code/response_code.rs

@@ -23,18 +23,6 @@ pub enum RemotingSysResponseCode {
     TransactionFailed = 4,
 }
 
-/*impl Into<i32> for RemotingSysResponseCode {
-    fn into(self) -> i32 {
-        match self {
-            RemotingSysResponseCode::Success => 0,
-            RemotingSysResponseCode::SystemError => 1,
-            RemotingSysResponseCode::SystemBusy => 2,
-            RemotingSysResponseCode::RequestCodeNotSupported => 3,
-            RemotingSysResponseCode::TransactionFailed => 4,
-        }
-    }
-}*/
-
 impl From<RemotingSysResponseCode> for i32 {
     fn from(value: RemotingSysResponseCode) -> Self {
         value as i32

rocketmq-remoting/src/protocol/header.rs

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

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

@@ -200,3 +200,38 @@ impl FromMap for DeleteKVConfigRequestHeader {
         })
     }
 }
+
+pub struct GetKVListByNamespaceRequestHeader {
+    pub namespace: String,
+}
+
+impl GetKVListByNamespaceRequestHeader {
+    const NAMESPACE: &'static str = "namespace";
+
+    pub fn new(namespace: impl Into<String>) -> Self {
+        Self {
+            namespace: namespace.into(),
+        }
+    }
+}
+
+impl CommandCustomHeader for GetKVListByNamespaceRequestHeader {
+    fn to_map(&self) -> Option<HashMap<String, String>> {
+        Some(HashMap::from([(
+            GetKVListByNamespaceRequestHeader::NAMESPACE.to_string(),
+            self.namespace.clone(),
+        )]))
+    }
+}
+
+impl FromMap for GetKVListByNamespaceRequestHeader {
+    type Target = GetKVListByNamespaceRequestHeader;
+
+    fn from(map: &HashMap<String, String>) -> Option<Self::Target> {
+        Some(GetKVListByNamespaceRequestHeader {
+            namespace: map
+                .get(GetKVListByNamespaceRequestHeader::NAMESPACE)
+                .cloned()?,
+        })
+    }
+}