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

Author: mxsm

Cargo.toml

@@ -1,10 +1,10 @@
 [workspace]
 members = [
+    "rocketmq",
     "rocketmq-common",
+    "rocketmq-macros",
     "rocketmq-namesrv",
-    "rocketmq-remoting",
-    "rocketmq"
-]
+    "rocketmq-remoting"]
 resolver = "2"
 
 [workspace.package]

rocketmq-macros/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "rocketmq-macros"
+version = "0.1.0"
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+repository.workspace = true
+license.workspace = true
+keywords.workspace = true
+readme.workspace = true
+description.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]

rocketmq-macros/src/lib.rs

@@ -0,0 +1,14 @@
+pub fn add(left: usize, right: usize) -> usize {
+    left + right
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn it_works() {
+        let result = add(2, 2);
+        assert_eq!(result, 4);
+    }
+}

rocketmq-namesrv/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "namesrv"
+name = "rocketmq-namesrv"
 version = "0.1.0"
 authors.workspace = true
 edition.workspace = true

rocketmq-namesrv/README.md

@@ -16,7 +16,7 @@ Feature list:
 
 | Feature                                | request code | Support        | remark |
 | -------------------------------------- | ------------ | -------------- | ------ |
-| Put KV Config                          | 100          | :broken_heart: |        |
+| Put KV Config                          | 100          | :heart: | ISSUE #40 |
 | Get KV Config                          | 101          | :broken_heart: |        |
 | Delete KV Config                       | 102          | :broken_heart: |        |
 | Get kv list by namespace               | 219          | :broken_heart: |        |

rocketmq-namesrv/src/bin/bootstrap_server.rs

@@ -18,11 +18,11 @@
 use std::{collections::HashMap, path::PathBuf, sync::Arc};
 
 use clap::Parser;
-use namesrv::{
+use rocketmq_common::common::namesrv::namesrv_config::NamesrvConfig;
+use rocketmq_namesrv::{
     processor::{default_request_processor::DefaultRequestProcessor, ClientRequestProcessor},
     KVConfigManager, RouteInfoManager,
 };
-use rocketmq_common::common::namesrv::namesrv_config::NamesrvConfig;
 use rocketmq_remoting::{
     code::request_code::RequestCode,
     runtime::{processor::RequestProcessor, server},
@@ -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 {
@@ -56,25 +60,55 @@ impl RequestProcessor for DefaultRequestProcessor {
             Some(RequestCode::GetBrokerClusterInfo) => {
                 self.process_get_broker_cluster_info(request)
             }
+            Some(RequestCode::PutKvConfig) => self.put_kv_config(request),
             _ => RemotingCommand::create_response_command_with_code(
                 RemotingSysResponseCode::SystemError,
             ),
         }
     }
 }
 
+///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()?,
+        })
+    }
+}