rewrite decode

Signed-off-by: junxiang Mu <1948535941@qq.com>

common/protos/src/generated/proto_gen/node_service.rs

@@ -293,8 +293,8 @@ pub struct WalkDirRequest {
     /// indicate which one in the disks
     #[prost(string, tag = "1")]
     pub disk: ::prost::alloc::string::String,
-    #[prost(string, tag = "2")]
-    pub walk_dir_options: ::prost::alloc::string::String,
+    #[prost(bytes = "vec", tag = "2")]
+    pub walk_dir_options: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct WalkDirResponse {

common/protos/src/node.proto

@@ -204,7 +204,7 @@ message ListDirResponse {
 
 message WalkDirRequest {
   string disk = 1;  // indicate which one in the disks
-  string walk_dir_options = 2;
+  bytes walk_dir_options = 2;
 }
 
 message WalkDirResponse {

e2e_test/src/reliant/node_interact_test.rs

@@ -11,8 +11,8 @@ use protos::{
         ListVolumesRequest, LocalStorageInfoRequest, MakeVolumeRequest, PingRequest, PingResponse, ReadAllRequest,
     },
 };
-use rmp_serde::Deserializer;
-use serde::Deserialize;
+use rmp_serde::{Deserializer, Serializer};
+use serde::{Deserialize, Serialize};
 use std::{error::Error, io::Cursor};
 use tokio::io::AsyncWrite;
 use tokio::spawn;
@@ -107,11 +107,12 @@ async fn walk_dir() -> Result<(), Box<dyn Error>> {
         ..Default::default()
     };
     let (rd, mut wr) = tokio::io::duplex(1024);
-    let walk_dir_options = serde_json::to_string(&opts)?;
+    let mut buf = Vec::new();
+    opts.serialize(&mut Serializer::new(&mut buf))?;
     let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
     let request = Request::new(WalkDirRequest {
         disk: "/home/dandan/code/rust/s3-rustfs/target/debug/data".to_string(),
-        walk_dir_options,
+        walk_dir_options: buf,
     });
     let mut response = client.walk_dir(request).await?.into_inner();
 

ecstore/src/disk/remote.rs

@@ -10,6 +10,8 @@ use protos::{
         StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
     },
 };
+use rmp_serde::Serializer;
+use serde::Serialize;
 use tokio::{
     io::AsyncWrite,
     sync::mpsc::{self, Sender},
@@ -356,13 +358,14 @@ impl DiskAPI for RemoteDisk {
         info!("walk_dir");
         let mut wr = wr;
         let mut out = MetacacheWriter::new(&mut wr);
-        let walk_dir_options = serde_json::to_string(&opts)?;
+        let mut buf = Vec::new();
+        opts.serialize(&mut Serializer::new(&mut buf))?;
         let mut client = node_service_time_out_client(&self.addr)
             .await
             .map_err(|err| Error::from_string(format!("can not get client, err: {}", err)))?;
         let request = Request::new(WalkDirRequest {
             disk: self.endpoint.to_string(),
-            walk_dir_options,
+            walk_dir_options: buf,
         });
         let mut response = client.walk_dir(request).await?.into_inner();
 

rustfs/src/grpc.rs

@@ -740,7 +740,8 @@ impl Node for NodeService {
         let request = request.into_inner();
         let (tx, rx) = mpsc::channel(128);
         if let Some(disk) = self.find_disk(&request.disk).await {
-            let opts = match serde_json::from_str::<WalkDirOptions>(&request.walk_dir_options) {
+            let mut buf = Deserializer::new(Cursor::new(request.walk_dir_options));
+            let opts = match Deserialize::deserialize(&mut buf) {
                 Ok(options) => options,
                 Err(_) => {
                     return Err(Status::invalid_argument("invalid WalkDirOptions"));