[ISSUE mxsm#703]🚀Implement RpcClientUtils#encode_body method🚀 (mxsm#707)

Author: mxsm

rocketmq-broker/src/offset/manager/consumer_offset_manager.rs

@@ -223,7 +223,3 @@ struct ConsumerOffsetWrapper {
     #[serde(skip)]
     version_change_counter: Arc<AtomicI64>,
 }
-
-impl RemotingSerializable for ConsumerOffsetWrapper {
-    type Output = Self;
-}

rocketmq-broker/src/out_api/broker_outer_api.rs

@@ -20,6 +20,7 @@ use dns_lookup::lookup_host;
 use rocketmq_common::common::broker::broker_config::BrokerIdentity;
 use rocketmq_common::common::config::TopicConfig;
 use rocketmq_common::utils::crc32_utils;
+use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use rocketmq_remoting::clients::rocketmq_default_impl::RocketmqDefaultClient;
 use rocketmq_remoting::clients::RemotingClient;
 use rocketmq_remoting::code::request_code::RequestCode;
@@ -209,7 +210,7 @@ impl BrokerOuterAPI {
                         result.master_addr = header.master_addr.clone().unwrap_or("".to_string());
                     }
                     if let Some(body) = response.body() {
-                        result.kv_table = KVTable::decode(body.as_ref());
+                        result.kv_table = SerdeJsonUtils::decode::<KVTable>(body.as_ref());
                     }
                     Some(result)
                 }

rocketmq-broker/src/processor/client_manage_processor.rs

@@ -24,12 +24,12 @@ use rocketmq_common::common::mix_all;
 use rocketmq_common::common::mix_all::IS_SUB_CHANGE;
 use rocketmq_common::common::mix_all::IS_SUPPORT_HEART_BEAT_V2;
 use rocketmq_common::common::sys_flag::topic_sys_flag;
+use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::protocol::header::unregister_client_request_header::UnregisterClientRequestHeader;
 use rocketmq_remoting::protocol::heartbeat::consume_type::ConsumeType;
 use rocketmq_remoting::protocol::heartbeat::heartbeat_data::HeartbeatData;
 use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
-use rocketmq_remoting::protocol::RemotingSerializable;
 use rocketmq_remoting::runtime::server::ConnectionHandlerContext;
 use rocketmq_store::log_file::MessageStore;
 use tracing::info;
@@ -129,8 +129,9 @@ where
         ctx: ConnectionHandlerContext<'_>,
         request: RemotingCommand,
     ) -> Option<RemotingCommand> {
-        let heartbeat_data =
-            HeartbeatData::decode(request.body().as_ref().map(|v| v.as_ref()).unwrap());
+        let heartbeat_data = SerdeJsonUtils::decode::<HeartbeatData>(
+            request.body().as_ref().map(|v| v.as_ref()).unwrap(),
+        );
         let client_channel_info = ClientChannelInfo::new(
             ctx.as_ref().connection().channel().clone(),
             heartbeat_data.client_id.clone(),

rocketmq-broker/src/subscription/manager/subscription_group_manager.rs

@@ -204,7 +204,3 @@ impl SubscriptionGroupWrapper {
         &self.forbidden_table
     }
 }
-
-impl RemotingSerializable for SubscriptionGroupWrapper {
-    type Output = Self;
-}

rocketmq-common/src/utils.rs

@@ -21,5 +21,6 @@ pub mod file_utils;
 pub mod message_utils;
 pub mod parse_config_file;
 pub mod queue_type_utils;
+pub mod serde_json_utils;
 pub mod time_utils;
 pub mod util_all;

rocketmq-common/src/utils/serde_json_utils.rs

@@ -19,6 +19,13 @@ use crate::error::SerdeJsonError;
 pub struct SerdeJsonUtils;
 
 impl SerdeJsonUtils {
+    pub fn decode<T>(bytes: &[u8]) -> T
+    where
+        T: serde::de::DeserializeOwned,
+    {
+        serde_json::from_slice::<T>(bytes).unwrap()
+    }
+
     pub fn from_json<T>(json: &str) -> Result<T, SerdeJsonError>
     where
         T: serde::de::DeserializeOwned,

rocketmq-namesrv/src/kvconfig.rs

@@ -17,7 +17,6 @@
 
 use std::collections::HashMap;
 
-use rocketmq_remoting::protocol::RemotingSerializable;
 use serde::Deserialize;
 use serde::Serialize;
 
@@ -45,7 +44,3 @@ impl KVConfigSerializeWrapper {
         }
     }
 }
-
-impl RemotingSerializable for KVConfigSerializeWrapper {
-    type Output = Self;
-}

rocketmq-namesrv/src/kvconfig/kvconfig_mananger.rs

@@ -18,6 +18,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use rocketmq_common::common::namesrv::namesrv_config::NamesrvConfig;
+use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use rocketmq_common::FileUtils;
 use rocketmq_remoting::protocol::body::kv_table::KVTable;
 use rocketmq_remoting::protocol::RemotingSerializable;
@@ -75,7 +76,7 @@ impl KVConfigManager {
     pub fn load(&mut self) {
         let result = FileUtils::file_to_string(self.namesrv_config.kv_config_path.as_str());
         if let Ok(content) = result {
-            let wrapper = KVConfigSerializeWrapper::decode(content.as_bytes());
+            let wrapper = SerdeJsonUtils::decode::<KVConfigSerializeWrapper>(content.as_bytes());
             if let Some(ref config_table) = wrapper.config_table {
                 for (namespace, config) in config_table {
                     self.config_table.insert(namespace.clone(), config.clone());

rocketmq-namesrv/src/processor/default_request_processor.rs

@@ -22,6 +22,7 @@ use bytes::Bytes;
 use rocketmq_common::common::mix_all;
 use rocketmq_common::common::mq_version::RocketMqVersion;
 use rocketmq_common::common::namesrv::namesrv_config::NamesrvConfig;
+use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use rocketmq_common::CRC32Utils;
 use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::code::response_code::RemotingSysResponseCode;
@@ -176,8 +177,9 @@ impl DefaultRequestProcessor {
         let request_header = request
             .decode_command_custom_header::<QueryDataVersionRequestHeader>()
             .unwrap();
-        let data_version =
-            DataVersion::decode(request.body().as_ref().map(|v| v.as_ref()).unwrap());
+        let data_version = SerdeJsonUtils::decode::<DataVersion>(
+            request.body().as_ref().map(|v| v.as_ref()).unwrap(),
+        );
         let changed = self
             .route_info_manager
             .read()
@@ -407,7 +409,7 @@ impl DefaultRequestProcessor {
             .decode_command_custom_header::<RegisterTopicRequestHeader>()
             .unwrap();
         if let Some(ref body) = request.body() {
-            let topic_route_data = TopicRouteData::decode(body);
+            let topic_route_data = SerdeJsonUtils::decode::<TopicRouteData>(body);
             if !topic_route_data.queue_datas.is_empty() {
                 self.route_info_manager
                     .write()
@@ -514,7 +516,9 @@ fn extract_register_topic_config_from_request(
     request: &RemotingCommand,
 ) -> TopicConfigAndMappingSerializeWrapper {
     if let Some(body_inner) = request.body() {
-        return TopicConfigAndMappingSerializeWrapper::decode(body_inner.iter().as_slice());
+        return SerdeJsonUtils::decode::<TopicConfigAndMappingSerializeWrapper>(
+            body_inner.iter().as_slice(),
+        );
     }
     TopicConfigAndMappingSerializeWrapper::default()
 }

rocketmq-remoting/src/protocol.rs

@@ -24,8 +24,8 @@ use std::time::SystemTime;
 
 use rocketmq_common::common::mix_all;
 use rocketmq_common::common::topic::TopicValidator;
+use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use rocketmq_common::utils::time_utils;
-use serde::de;
 use serde::ser::SerializeStruct;
 use serde::Deserialize;
 use serde::Serialize;
@@ -222,10 +222,6 @@ impl<'de> Deserialize<'de> for DataVersion {
     }
 }
 
-impl RemotingSerializable for DataVersion {
-    type Output = DataVersion;
-}
-
 impl Clone for DataVersion {
     fn clone(&self) -> Self {
         DataVersion {
@@ -326,44 +322,24 @@ impl Display for DataVersion {
 
 /// A trait for types that can be deserialized from a byte vector.
 pub trait RemotingSerializable {
-    /// The output type after deserialization.
-    type Output;
-
-    /// Decode a byte vector into the corresponding type.
-    ///
-    /// # Arguments
-    ///
-    /// * `bytes` - The byte vector to be deserialized.
-    ///
-    /// # Returns
-    ///
-    /// The deserialized output of type `Self::Output`.
-    fn decode<'a>(bytes: &'a [u8]) -> Self::Output
-    where
-        Self::Output: de::Deserialize<'a>,
-    {
-        serde_json::from_slice::<Self::Output>(bytes).unwrap()
-    }
+    fn encode(&self) -> Vec<u8>;
+    fn to_json(&self) -> String;
+    fn to_json_pretty(&self) -> String;
+}
 
-    fn encode(&self) -> Vec<u8>
-    where
-        Self: Serialize,
-    {
-        serde_json::to_vec(self).unwrap()
+pub trait JsonSerializable: Serialize + RemotingSerializable {}
+
+impl<T: Serialize> RemotingSerializable for T {
+    fn encode(&self) -> Vec<u8> {
+        SerdeJsonUtils::to_json_vec(self).unwrap()
     }
 
-    fn to_json(&self) -> String
-    where
-        Self: Serialize,
-    {
-        serde_json::to_string(self).unwrap()
+    fn to_json(&self) -> String {
+        SerdeJsonUtils::to_json(self).unwrap()
     }
 
-    fn to_json_pretty(&self) -> String
-    where
-        Self: Serialize,
-    {
-        serde_json::to_string_pretty(self).unwrap()
+    fn to_json_pretty(&self) -> String {
+        SerdeJsonUtils::to_json_pretty(self).unwrap()
     }
 }
 

rocketmq-remoting/src/protocol/body/broker_body/broker_member_group.rs

@@ -19,8 +19,6 @@ use std::collections::HashMap;
 use serde::Deserialize;
 use serde::Serialize;
 
-use crate::protocol::RemotingSerializable;
-
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct BrokerMemberGroup {
@@ -43,16 +41,8 @@ impl BrokerMemberGroup {
     }
 }
 
-impl RemotingSerializable for BrokerMemberGroup {
-    type Output = BrokerMemberGroup;
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct GetBrokerMemberGroupResponseBody {
     pub broker_member_group: Option<BrokerMemberGroup>,
 }
-
-impl RemotingSerializable for GetBrokerMemberGroupResponseBody {
-    type Output = GetBrokerMemberGroupResponseBody;
-}

rocketmq-remoting/src/protocol/body/broker_body/cluster_info.rs

@@ -22,7 +22,6 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use crate::protocol::route::route_data_view::BrokerData;
-use crate::protocol::RemotingSerializable;
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct ClusterInfo {
@@ -33,18 +32,6 @@ pub struct ClusterInfo {
     cluster_addr_table: Option<HashMap<String, HashSet<String>>>,
 }
 
-impl RemotingSerializable for ClusterInfo {
-    type Output = ClusterInfo;
-
-    /*fn decode(bytes: &[u8]) -> ClusterInfo {
-        serde_json::from_slice::<Self::Output>(bytes).unwrap()
-    }
-
-    fn encode(&self, _compress: bool) -> Vec<u8> {
-        serde_json::to_vec(self).unwrap()
-    }*/
-}
-
 impl ClusterInfo {
     pub fn new(
         broker_addr_table: Option<HashMap<String, BrokerData>>,

rocketmq-remoting/src/protocol/body/broker_body/register_broker_body.rs

@@ -17,6 +17,7 @@
 
 use bytes::Bytes;
 use rocketmq_common::common::mq_version::RocketMqVersion;
+use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use serde::Deserialize;
 use serde::Serialize;
 
@@ -66,20 +67,8 @@ impl RegisterBrokerBody {
         _broker_version: RocketMqVersion,
     ) -> RegisterBrokerBody {
         if !compressed {
-            return <RegisterBrokerBody as RemotingSerializable>::decode(bytes.iter().as_slice());
+            return SerdeJsonUtils::decode::<RegisterBrokerBody>(bytes.iter().as_slice());
         }
         todo!()
     }
 }
-
-impl RemotingSerializable for RegisterBrokerBody {
-    type Output = RegisterBrokerBody;
-
-    /* fn decode(bytes: &[u8]) -> Self::Output {
-        serde_json::from_slice::<Self::Output>(bytes).unwrap()
-    }
-
-    fn encode(&self, _compress: bool) -> Vec<u8> {
-        todo!()
-    }*/
-}

rocketmq-remoting/src/protocol/body/get_consumer_listby_group_response_body.rs

@@ -17,14 +17,8 @@
 use serde::Deserialize;
 use serde::Serialize;
 
-use crate::protocol::RemotingSerializable;
-
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct GetConsumerListByGroupResponseBody {
     pub consumer_id_list: Vec<String>,
 }
-
-impl RemotingSerializable for GetConsumerListByGroupResponseBody {
-    type Output = ();
-}

rocketmq-remoting/src/protocol/body/kv_table.rs

@@ -20,13 +20,7 @@ use std::collections::HashMap;
 use serde::Deserialize;
 use serde::Serialize;
 
-use crate::protocol::RemotingSerializable;
-
 #[derive(Deserialize, Serialize, Debug, Clone, Default)]
 pub struct KVTable {
     pub table: HashMap<String, String>,
 }
-
-impl RemotingSerializable for KVTable {
-    type Output = Self;
-}

rocketmq-remoting/src/protocol/body/topic/topic_list.rs

@@ -17,19 +17,13 @@
 use serde::Deserialize;
 use serde::Serialize;
 
-use crate::protocol::RemotingSerializable;
-
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct TopicList {
     pub topic_list: Vec<String>,
     pub broker_addr: Option<String>,
 }
 
-impl RemotingSerializable for TopicList {
-    type Output = TopicList;
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

rocketmq-remoting/src/protocol/body/topic_info_wrapper.rs

@@ -22,7 +22,6 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use crate::protocol::DataVersion;
-use crate::protocol::RemotingSerializable;
 
 pub mod topic_config_wrapper;
 pub mod topic_queue_wrapper;
@@ -74,7 +73,3 @@ impl TopicConfigSerializeWrapper {
         self.data_version = data_version;
     }
 }
-
-impl RemotingSerializable for TopicConfigSerializeWrapper {
-    type Output = Self;
-}

rocketmq-remoting/src/protocol/body/topic_info_wrapper/topic_config_wrapper.rs

@@ -23,7 +23,6 @@ use serde::Serialize;
 use crate::protocol::static_topic::topic_queue_info::TopicQueueMappingInfo;
 use crate::protocol::static_topic::topic_queue_mapping_detail::TopicQueueMappingDetail;
 use crate::protocol::DataVersion;
-use crate::protocol::RemotingSerializable;
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct TopicConfigAndMappingSerializeWrapper {
@@ -75,10 +74,6 @@ impl Default for TopicConfigAndMappingSerializeWrapper {
     }
 }
 
-impl RemotingSerializable for TopicConfigAndMappingSerializeWrapper {
-    type Output = TopicConfigAndMappingSerializeWrapper;
-}
-
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;