[ISSUE #1232]Optimize DefaultRequestProcessor code

mxsm · mxsm · commit b0d2ea423ca4 · 2024-11-19T22:12:53.000+08:00
diff --git a/rocketmq-namesrv/src/bootstrap.rs b/rocketmq-namesrv/src/bootstrap.rs
@@ -101,8 +101,8 @@ impl NameServerRuntime {
                 Duration::from_secs(5),
             );
         NameServerRequestProcessor {
-            client_request_processor: Arc::new(client_request_processor),
-            default_request_processor: Arc::new(default_request_processor),
+            client_request_processor: ArcMut::new(client_request_processor),
+            default_request_processor: ArcMut::new(default_request_processor),
         }
     }
 }
diff --git a/rocketmq-namesrv/src/processor.rs b/rocketmq-namesrv/src/processor.rs
@@ -15,14 +15,13 @@
  * limitations under the License.
  */
 
-use std::sync::Arc;
-
 use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::net::channel::Channel;
 use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
 use rocketmq_remoting::runtime::connection_handler_context::ConnectionHandlerContext;
 use rocketmq_remoting::runtime::processor::RequestProcessor;
 use rocketmq_remoting::Result;
+use rocketmq_rust::ArcMut;
 use tracing::info;
 
 pub use self::client_request_processor::ClientRequestProcessor;
@@ -31,18 +30,10 @@ use crate::processor::default_request_processor::DefaultRequestProcessor;
 mod client_request_processor;
 pub mod default_request_processor;
 
+#[derive(Clone)]
 pub struct NameServerRequestProcessor {
-    pub(crate) client_request_processor: Arc<ClientRequestProcessor>,
-    pub(crate) default_request_processor: Arc<DefaultRequestProcessor>,
-}
-
-impl Clone for NameServerRequestProcessor {
-    fn clone(&self) -> Self {
-        Self {
-            client_request_processor: self.client_request_processor.clone(),
-            default_request_processor: self.default_request_processor.clone(),
-        }
-    }
+    pub(crate) client_request_processor: ArcMut<ClientRequestProcessor>,
+    pub(crate) default_request_processor: ArcMut<DefaultRequestProcessor>,
 }
 
 impl RequestProcessor for NameServerRequestProcessor {
@@ -53,14 +44,16 @@ impl RequestProcessor for NameServerRequestProcessor {
         request: RemotingCommand,
     ) -> Result<Option<RemotingCommand>> {
         let request_code = RequestCode::from(request.code());
-        info!("process_request: {:?}", request_code);
+        info!("Name server Received request code: {:?}", request_code);
         let result = match request_code {
-            RequestCode::GetRouteinfoByTopic => self
-                .client_request_processor
-                .process_request(channel, ctx, request),
-            _ => self
-                .default_request_processor
-                .process_request(channel, ctx, request),
+            RequestCode::GetRouteinfoByTopic => {
+                self.client_request_processor
+                    .process_request(channel, ctx, request_code, request)
+            }
+            _ => {
+                self.default_request_processor
+                    .process_request(channel, ctx, request_code, request)
+            }
         };
         Ok(result)
     }
diff --git a/rocketmq-namesrv/src/processor/client_request_processor.rs b/rocketmq-namesrv/src/processor/client_request_processor.rs
@@ -23,6 +23,7 @@ use std::time::Duration;
 use rocketmq_common::common::namesrv::namesrv_config::NamesrvConfig;
 use rocketmq_common::common::FAQUrl;
 use rocketmq_common::TimeUtils;
+use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::code::response_code::RemotingSysResponseCode;
 use rocketmq_remoting::code::response_code::ResponseCode;
 use rocketmq_remoting::net::channel::Channel;
@@ -118,9 +119,10 @@ impl ClientRequestProcessor {
 
 impl ClientRequestProcessor {
     pub fn process_request(
-        &self,
+        &mut self,
         _channel: Channel,
         _ctx: ConnectionHandlerContext,
+        _request_code: RequestCode,
         request: RemotingCommand,
     ) -> Option<RemotingCommand> {
         Some(self.get_route_info_by_topic(request))
diff --git a/rocketmq-namesrv/src/processor/default_request_processor.rs b/rocketmq-namesrv/src/processor/default_request_processor.rs
@@ -54,61 +54,51 @@ use rocketmq_remoting::protocol::route::topic_route_data::TopicRouteData;
 use rocketmq_remoting::protocol::DataVersion;
 use rocketmq_remoting::protocol::RemotingSerializable;
 use rocketmq_remoting::runtime::connection_handler_context::ConnectionHandlerContext;
-use tracing::info;
 use tracing::warn;
 
 use crate::route::route_info_manager::RouteInfoManager;
 use crate::KVConfigManager;
 
-#[derive(Clone)]
 pub struct DefaultRequestProcessor {
     route_info_manager: Arc<parking_lot::RwLock<RouteInfoManager>>,
     kvconfig_manager: Arc<parking_lot::RwLock<KVConfigManager>>,
 }
 
 impl DefaultRequestProcessor {
     pub fn process_request(
-        &self,
+        &mut self,
         channel: Channel,
         _ctx: ConnectionHandlerContext,
+        request_code: RequestCode,
         request: RemotingCommand,
     ) -> Option<RemotingCommand> {
-        let code = request.code();
-        let broker_request_code = RequestCode::value_of(code);
-        info!(
-            "Received request code:{}-{:?}",
-            code,
-            broker_request_code.as_ref()
-        );
-        let response = match broker_request_code {
-            Some(RequestCode::PutKvConfig) => self.put_kv_config(request),
-            Some(RequestCode::GetKvConfig) => self.get_kv_config(request),
-            Some(RequestCode::DeleteKvConfig) => self.delete_kv_config(request),
-            Some(RequestCode::QueryDataVersion) => self.query_broker_topic_config(request),
+        let response = match request_code {
+            RequestCode::PutKvConfig => self.put_kv_config(request),
+            RequestCode::GetKvConfig => self.get_kv_config(request),
+            RequestCode::DeleteKvConfig => self.delete_kv_config(request),
+            RequestCode::QueryDataVersion => self.query_broker_topic_config(request),
             //handle register broker
-            Some(RequestCode::RegisterBroker) => {
+            RequestCode::RegisterBroker => {
                 self.process_register_broker(channel.remote_address(), request)
             }
-            Some(RequestCode::UnregisterBroker) => self.process_unregister_broker(request),
-            Some(RequestCode::BrokerHeartbeat) => self.process_broker_heartbeat(request),
-            Some(RequestCode::GetBrokerMemberGroup) => self.get_broker_member_group(request),
+            RequestCode::UnregisterBroker => self.process_unregister_broker(request),
+            RequestCode::BrokerHeartbeat => self.process_broker_heartbeat(request),
+            RequestCode::GetBrokerMemberGroup => self.get_broker_member_group(request),
             //handle get broker cluster info
-            Some(RequestCode::GetBrokerClusterInfo) => self.get_broker_cluster_info(request),
-            Some(RequestCode::WipeWritePermOfBroker) => self.wipe_write_perm_of_broker(request),
-            Some(RequestCode::AddWritePermOfBroker) => self.add_write_perm_of_broker(request),
-            Some(RequestCode::GetAllTopicListFromNameserver) => {
+            RequestCode::GetBrokerClusterInfo => self.get_broker_cluster_info(request),
+            RequestCode::WipeWritePermOfBroker => self.wipe_write_perm_of_broker(request),
+            RequestCode::AddWritePermOfBroker => self.add_write_perm_of_broker(request),
+            RequestCode::GetAllTopicListFromNameserver => {
                 self.get_all_topic_list_from_nameserver(request)
             }
-            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),
-            Some(RequestCode::GetTopicsByCluster) => self.get_topics_by_cluster(request),
-            Some(RequestCode::GetSystemTopicListFromNs) => {
-                self.get_system_topic_list_from_ns(request)
-            }
-            Some(RequestCode::GetUnitTopicList) => self.get_unit_topic_list(request),
-            Some(RequestCode::GetHasUnitSubTopicList) => self.get_has_unit_sub_topic_list(request),
-            Some(RequestCode::GetHasUnitSubUnunitTopicList) => {
+            RequestCode::DeleteTopicInNamesrv => self.delete_topic_in_name_srv(request),
+            RequestCode::RegisterTopicInNamesrv => self.register_topic_to_name_srv(request),
+            RequestCode::GetKvlistByNamespace => self.get_kv_list_by_namespace(request),
+            RequestCode::GetTopicsByCluster => self.get_topics_by_cluster(request),
+            RequestCode::GetSystemTopicListFromNs => self.get_system_topic_list_from_ns(request),
+            RequestCode::GetUnitTopicList => self.get_unit_topic_list(request),
+            RequestCode::GetHasUnitSubTopicList => self.get_has_unit_sub_topic_list(request),
+            RequestCode::GetHasUnitSubUnunitTopicList => {
                 self.get_has_unit_sub_un_unit_topic_list(request)
             }
             _ => RemotingCommand::create_response_command_with_code(

Original file line number	Diff line number	Diff line change
`@@ -101,8 +101,8 @@ impl NameServerRuntime {`
`101`	`101`	`Duration::from_secs(5),`
`102`	`102`	`);`
`103`	`103`	`NameServerRequestProcessor {`
`104`		`- client_request_processor: Arc::new(client_request_processor),`
`105`		`- default_request_processor: Arc::new(default_request_processor),`
	`104`	`+ client_request_processor: ArcMut::new(client_request_processor),`
	`105`	`+ default_request_processor: ArcMut::new(default_request_processor),`
`106`	`106`	`}`
`107`	`107`	`}`
`108`	`108`	`}`