[ISSUE #288]🎨Optimize ArcProcessorTable code (#289)

* [ISSUE #288]🎨Optimize ArcProcessorTable code

* remove useless file

Author: TeslaRustor

rocketmq-namesrv/src/bin/bootstrap_server.rs

@@ -28,7 +28,7 @@ use rocketmq_namesrv::{
 };
 use rocketmq_remoting::{
     code::request_code::RequestCode,
-    runtime::{processor::RequestProcessor, server},
+    runtime::{processor::RequestProcessor, server, ArcProcessorTable},
 };
 use rocketmq_rust::rocketmq;
 use tokio::{net::TcpListener, sync::broadcast, task::JoinHandle};
@@ -62,7 +62,7 @@ async fn main() -> anyhow::Result<()> {
         listener,
         tokio::signal::ctrl_c(),
         Arc::new(Box::new(default_request_processor)),
-        Arc::new(processor_table),
+        processor_table,
         Some(notify_conn_disconnect),
     )
     .await;
@@ -71,7 +71,7 @@ async fn main() -> anyhow::Result<()> {
 }
 
 type InitProcessorsReturn = (
-    HashMap<i32, Box<dyn RequestProcessor + Send + Sync + 'static>>,
+    ArcProcessorTable,
     DefaultRequestProcessor,
     ScheduledExecutorService,
     JoinHandle<()>,
@@ -86,16 +86,16 @@ fn init_processors(
     let route_info_manager_inner = Arc::new(parking_lot::RwLock::new(route_info_manager));
     let handle = RouteInfoManager::start(route_info_manager_inner.clone(), receiver);
     let kvconfig_manager_inner = Arc::new(parking_lot::RwLock::new(kvconfig_manager));
-    let mut processors: HashMap<i32, Box<dyn RequestProcessor + Send + Sync + 'static>> =
+    let mut processors: HashMap<i32, Arc<Box<dyn RequestProcessor + Send + Sync + 'static>>> =
         HashMap::new();
 
     processors.insert(
         RequestCode::GetRouteinfoByTopic.to_i32(),
-        Box::new(ClientRequestProcessor::new(
+        Arc::new(Box::new(ClientRequestProcessor::new(
             route_info_manager_inner.clone(),
             namesrv_config.clone(),
             kvconfig_manager_inner.clone(),
-        )),
+        ))),
     );
     let scheduled_executor_service = ScheduledExecutorService::new_with_config(
         1,

rocketmq-remoting/src/runtime.rs

@@ -17,7 +17,6 @@
 
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
-use rocketmq_common::{common::Pair, TokioExecutorService};
 use tokio::{time, time::timeout};
 
 use crate::{
@@ -34,7 +33,7 @@ pub mod server;
 
 pub type ArcDefaultRequestProcessor = Arc<Box<dyn RequestProcessor + Send + Sync + 'static>>;
 
-pub type ArcProcessorTable = Arc<HashMap<i32, Box<dyn RequestProcessor + Sync + Send + 'static>>>;
+pub type ArcProcessorTable = HashMap<i32, ArcDefaultRequestProcessor>;
 
 pub trait RPCHook: Send + Sync + 'static {
     fn do_before_request(&self, remote_addr: &str, request: &RemotingCommand);
@@ -58,13 +57,6 @@ pub struct ServiceBridge {
     pub(crate) processor_table: Option<ArcProcessorTable>,
     pub(crate) default_request_processor_pair: Option<ArcDefaultRequestProcessor>,
 
-    pub(crate) processor_table1: HashMap<
-        i32, /* request code */
-        Pair<Box<dyn RequestProcessor + Send + Sync>, TokioExecutorService>,
-    >,
-    pub(crate) default_request_processor_pair1:
-        Option<Pair<Box<dyn RequestProcessor + Send + Sync>, TokioExecutorService>>,
-
     pub(crate) rpc_hooks: Vec<Box<dyn RPCHook>>,
 }
 
@@ -74,10 +66,8 @@ impl ServiceBridge {
             semaphore_oneway: tokio::sync::Semaphore::new(1000),
             semaphore_async: tokio::sync::Semaphore::new(1000),
             response_table: HashMap::new(),
-            processor_table: Some(Arc::new(HashMap::new())),
+            processor_table: Some(HashMap::new()),
             default_request_processor_pair: None,
-            processor_table1: Default::default(),
-            default_request_processor_pair1: None,
             rpc_hooks: Vec::new(),
         }
     }