refactor(meta-service): add MetaGrpcReq, as a replacement of MetaGrpcReadReq and MetaGrpcWriteReq.

There is no need to distinguish read/write at the request level.

Author: drmingdrmer

src/meta/client/src/grpc_action.rs

@@ -59,6 +59,33 @@ pub enum MetaGrpcReadReq {
     ListKV(ListKVReq), // since 2022-05-23
 }
 
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
+pub enum MetaGrpcReq {
+    UpsertKV(UpsertKVReq),
+
+    GetKV(GetKVReq),
+    MGetKV(MGetKVReq),
+    ListKV(ListKVReq),
+}
+
+impl From<MetaGrpcWriteReq> for MetaGrpcReq {
+    fn from(r: MetaGrpcWriteReq) -> Self {
+        match r {
+            MetaGrpcWriteReq::UpsertKV(x) => x.into(),
+        }
+    }
+}
+
+impl From<MetaGrpcReadReq> for MetaGrpcReq {
+    fn from(r: MetaGrpcReadReq) -> Self {
+        match r {
+            MetaGrpcReadReq::GetKV(x) => x.into(),
+            MetaGrpcReadReq::MGetKV(x) => x.into(),
+            MetaGrpcReadReq::ListKV(x) => x.into(),
+        }
+    }
+}
+
 /// Try convert tonic::Request<RaftRequest> to DoActionAction.
 impl TryInto<MetaGrpcWriteReq> for Request<RaftRequest> {
     type Error = tonic::Status;

src/meta/client/src/lib.rs

@@ -18,6 +18,7 @@ mod kv_api_impl;
 mod message;
 
 pub use grpc_action::MetaGrpcReadReq;
+pub use grpc_action::MetaGrpcReq;
 pub use grpc_action::MetaGrpcWriteReq;
 pub use grpc_action::RequestFor;
 pub use grpc_client::ClientHandle;

src/meta/service/src/api/grpc/grpc_service.rs

@@ -22,6 +22,7 @@ use common_base::base::tokio::sync::mpsc;
 use common_grpc::GrpcClaim;
 use common_grpc::GrpcToken;
 use common_meta_client::MetaGrpcReadReq;
+use common_meta_client::MetaGrpcReq;
 use common_meta_client::MetaGrpcWriteReq;
 use common_meta_types::protobuf::meta_service_server::MetaService;
 use common_meta_types::protobuf::ClientInfo;
@@ -154,13 +155,14 @@ impl MetaService for MetaServiceImpl {
 
         incr_meta_metrics_meta_recv_bytes(request.get_ref().encoded_len() as u64);
 
-        let action: MetaGrpcWriteReq = request.try_into()?;
+        let req: MetaGrpcWriteReq = request.try_into()?;
+        let req: MetaGrpcReq = req.into();
 
         add_meta_metrics_meta_request_inflights(1);
 
-        info!("Receive write_action: {:?}", action);
+        info!("Receive write_action: {:?}", req);
 
-        let body = self.action_handler.execute_write(action).await;
+        let body = self.action_handler.execute_kv_req(req).await;
 
         add_meta_metrics_meta_request_inflights(-1);
 
@@ -175,13 +177,14 @@ impl MetaService for MetaServiceImpl {
 
         incr_meta_metrics_meta_recv_bytes(request.get_ref().encoded_len() as u64);
 
-        let action: MetaGrpcReadReq = request.try_into()?;
+        let req: MetaGrpcReadReq = request.try_into()?;
+        let req: MetaGrpcReq = req.into();
 
         add_meta_metrics_meta_request_inflights(1);
 
-        info!("Receive read_action: {:?}", action);
+        info!("Receive read_action: {:?}", req);
 
-        let res = self.action_handler.execute_read(action).await;
+        let res = self.action_handler.execute_kv_req(req).await;
 
         add_meta_metrics_meta_request_inflights(-1);
 

src/meta/service/src/executor/action_handler.rs

@@ -15,11 +15,8 @@
 use std::sync::Arc;
 
 use common_meta_api::KVApi;
-use common_meta_client::MetaGrpcReadReq;
-use common_meta_client::MetaGrpcWriteReq;
-use common_meta_client::RequestFor;
+use common_meta_client::MetaGrpcReq;
 use common_meta_types::protobuf::RaftReply;
-use common_meta_types::MetaError;
 use common_meta_types::TxnReply;
 use common_meta_types::TxnRequest;
 
@@ -31,48 +28,34 @@ pub struct ActionHandler {
     pub(crate) meta_node: Arc<MetaNode>,
 }
 
-#[async_trait::async_trait]
-pub trait RequestHandler<T>: Sync + Send
-where T: RequestFor
-{
-    async fn handle(&self, req: T) -> Result<T::Reply, MetaError>;
-}
-
 impl ActionHandler {
     pub fn create(meta_node: Arc<MetaNode>) -> Self {
         ActionHandler { meta_node }
     }
 
-    pub async fn execute_write(&self, action: MetaGrpcWriteReq) -> RaftReply {
+    pub async fn execute_kv_req(&self, req: MetaGrpcReq) -> RaftReply {
         // To keep the code IDE-friendly, we manually expand the enum variants and dispatch them one by one
 
-        match action {
-            MetaGrpcWriteReq::UpsertKV(a) => {
-                let r = self.meta_node.upsert_kv(a).await;
-                incr_meta_metrics_meta_request_result(r.is_ok());
-                RaftReply::from(r)
+        match req {
+            MetaGrpcReq::UpsertKV(a) => {
+                let res = self.meta_node.upsert_kv(a).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
             }
-        }
-    }
-
-    pub async fn execute_read(&self, action: MetaGrpcReadReq) -> RaftReply {
-        // To keep the code IDE-friendly, we manually expand the enum variants and dispatch them one by one
-
-        match action {
-            MetaGrpcReadReq::GetKV(a) => {
-                let r = self.meta_node.get_kv(&a.key).await;
-                incr_meta_metrics_meta_request_result(r.is_ok());
-                RaftReply::from(r)
+            MetaGrpcReq::GetKV(a) => {
+                let res = self.meta_node.get_kv(&a.key).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
             }
-            MetaGrpcReadReq::MGetKV(a) => {
-                let r = self.meta_node.mget_kv(&a.keys).await;
-                incr_meta_metrics_meta_request_result(r.is_ok());
-                RaftReply::from(r)
+            MetaGrpcReq::MGetKV(a) => {
+                let res = self.meta_node.mget_kv(&a.keys).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
             }
-            MetaGrpcReadReq::ListKV(a) => {
-                let r = self.meta_node.prefix_list_kv(&a.prefix).await;
-                incr_meta_metrics_meta_request_result(r.is_ok());
-                RaftReply::from(r)
+            MetaGrpcReq::ListKV(a) => {
+                let res = self.meta_node.prefix_list_kv(&a.prefix).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
             }
         }
     }