[ISSUE #40]Support put kv config(request code 100)

Author: mxsm

rocketmq-namesrv/src/bin/bootstrap_server.rs

@@ -80,11 +80,14 @@ fn init_processors(
         RequestCode::GetRouteinfoByTopic.to_i32(),
         Box::new(ClientRequestProcessor::new(
             route_info_manager_inner.clone(),
-            namesrv_config,
+            namesrv_config.clone(),
             kvconfig_manager_inner.clone(),
         )),
     );
-    (processors, DefaultRequestProcessor::new())
+    (
+        processors,
+        DefaultRequestProcessor::new(namesrv_config.clone()),
+    )
 }
 
 #[derive(Parser, Debug)]

rocketmq-namesrv/src/kvconfig/kvconfig_mananger.rs

@@ -22,6 +22,7 @@ use tracing::info;
 
 use crate::kvconfig::KVConfigSerializeWrapper;
 
+#[derive(Debug)]
 pub struct KVConfigManager {
     pub(crate) config_table:
         HashMap<String /* Namespace */, HashMap<String /* Key */, String /* Value */>>,

rocketmq-namesrv/src/processor/default_request_processor.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use rocketmq_common::{
-    common::{mix_all, mq_version::RocketMqVersion},
+    common::{mix_all, mq_version::RocketMqVersion, namesrv::namesrv_config::NamesrvConfig},
     CRC32Utils,
 };
 use rocketmq_remoting::{
@@ -29,19 +29,23 @@ use rocketmq_remoting::{
             broker_body::register_broker_body::RegisterBrokerBody,
             topic_info_wrapper::topic_config_wrapper::TopicConfigAndMappingSerializeWrapper,
         },
-        header::broker_request_header::RegisterBrokerRequestHeader,
+        header::{
+            broker_request_header::RegisterBrokerRequestHeader,
+            namesrv::kv_config_request_header::PutKVConfigRequestHeader,
+        },
         remoting_command::RemotingCommand,
         RemotingSerializable,
     },
     runtime::processor::RequestProcessor,
 };
 use tracing::warn;
 
-use crate::route::route_info_manager::RouteInfoManager;
+use crate::{route::route_info_manager::RouteInfoManager, KVConfigManager};
 
 #[derive(Debug, Clone)]
 pub struct DefaultRequestProcessor {
     route_info_manager: Arc<parking_lot::RwLock<RouteInfoManager>>,
+    kvconfig_manager: Arc<parking_lot::RwLock<KVConfigManager>>,
 }
 
 impl RequestProcessor for DefaultRequestProcessor {
@@ -63,18 +67,47 @@ impl RequestProcessor for DefaultRequestProcessor {
     }
 }
 
+///implementation put KV config
+impl DefaultRequestProcessor {
+    fn put_kv_config(&mut self, request: RemotingCommand) -> RemotingCommand {
+        let request_header = request
+            .decode_command_custom_header::<PutKVConfigRequestHeader>()
+            .unwrap();
+        //check namespace and key, need?
+        if request_header.namespace.is_empty() || request_header.key.is_empty() {
+            return RemotingCommand::create_response_command_with_code(
+                RemotingSysResponseCode::SystemError,
+            )
+            .set_remark(Some(String::from("namespace or key is empty")));
+        }
+        self.kvconfig_manager.write().put_kv_config(
+            request_header.namespace.as_str(),
+            request_header.key.as_str(),
+            request_header.value.as_str(),
+        );
+        RemotingCommand::create_response_command()
+    }
+}
+
 #[allow(clippy::new_without_default)]
 impl DefaultRequestProcessor {
-    pub fn new() -> Self {
+    pub fn new(namesrv_config: NamesrvConfig) -> Self {
         Self {
             route_info_manager: Arc::new(parking_lot::RwLock::new(RouteInfoManager::new())),
+            kvconfig_manager: Arc::new(parking_lot::RwLock::new(KVConfigManager::new(
+                namesrv_config,
+            ))),
         }
     }
 
     pub(crate) fn new_with_route_info_manager(
         route_info_manager: Arc<parking_lot::RwLock<RouteInfoManager>>,
+        kvconfig_manager: Arc<parking_lot::RwLock<KVConfigManager>>,
     ) -> Self {
-        Self { route_info_manager }
+        Self {
+            route_info_manager,
+            kvconfig_manager,
+        }
     }
 }
 impl DefaultRequestProcessor {

rocketmq-remoting/src/protocol/command_custom_header.rs

@@ -23,7 +23,9 @@ pub trait CommandCustomHeader {
     /// Returns a `Result` indicating whether the fields are valid or not.  
     /// If the fields are valid, the `Ok` variant is returned with an empty `()` value.  
     /// If the fields are invalid, an `Err` variant is returned with an associated `Error` value.  
-    fn check_fields(&self) -> anyhow::Result<(), anyhow::Error>;
+    fn check_fields(&self) -> anyhow::Result<(), anyhow::Error> {
+        Ok(())
+    }
 
     /// Converts the implementing type to a map.  
     ///  

rocketmq-remoting/src/protocol/header.rs

@@ -18,3 +18,4 @@ pub mod broker;
 pub mod broker_request_header;
 pub mod client_request_header;
 pub mod kv_config_request_header;
+pub mod namesrv;

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

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

@@ -0,0 +1,82 @@
+/*
+ * 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)]
+pub struct PutKVConfigRequestHeader {
+    pub namespace: String,
+    pub key: String,
+    pub value: String,
+}
+
+impl PutKVConfigRequestHeader {
+    const NAMESPACE: &'static str = "namespace";
+    const KEY: &'static str = "key";
+    const VALUE: &'static str = "value";
+
+    /// Creates a new instance of `PutKVConfigRequestHeader`.
+    ///
+    /// # Arguments
+    ///
+    /// * `namespace` - The namespace.
+    /// * `key` - The key.
+    /// * `value` - The value.
+    pub fn new(
+        namespace: impl Into<String>,
+        key: impl Into<String>,
+        value: impl Into<String>,
+    ) -> Self {
+        Self {
+            namespace: namespace.into(),
+            key: key.into(),
+            value: value.into(),
+        }
+    }
+}
+
+impl CommandCustomHeader for PutKVConfigRequestHeader {
+    fn to_map(&self) -> Option<HashMap<String, String>> {
+        Some(HashMap::from([
+            (
+                PutKVConfigRequestHeader::NAMESPACE.to_string(),
+                self.namespace.clone(),
+            ),
+            (PutKVConfigRequestHeader::KEY.to_string(), self.key.clone()),
+            (
+                PutKVConfigRequestHeader::VALUE.to_string(),
+                self.value.clone(),
+            ),
+        ]))
+    }
+}
+
+impl FromMap for PutKVConfigRequestHeader {
+    type Target = PutKVConfigRequestHeader;
+
+    fn from(map: &HashMap<String, String>) -> Option<Self::Target> {
+        Some(PutKVConfigRequestHeader {
+            namespace: map.get(PutKVConfigRequestHeader::NAMESPACE).cloned()?,
+            key: map.get(PutKVConfigRequestHeader::KEY).cloned()?,
+            value: map.get(PutKVConfigRequestHeader::VALUE).cloned()?,
+        })
+    }
+}