diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 3cb6ec017..0b6da5aff 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1521,7 +1521,7 @@ impl DiskAPI for LocalDisk { // FIXME: TODO: io.writer TODO cancel #[tracing::instrument(level = "debug", skip(self, wr))] - async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result> { + async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { let volume_dir = self.get_bucket_path(&opts.bucket)?; if !skip_access_checks(&opts.bucket) { @@ -1560,7 +1560,7 @@ impl DiskAPI for LocalDisk { let mut current = opts.base_dir.clone(); self.scan_dir(&mut current, &opts, &mut out, &mut objs_returned).await?; - Ok(Vec::new()) + Ok(()) } // #[tracing::instrument(skip(self))] diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 000d0d0dd..c5f4fd9a9 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -211,7 +211,7 @@ impl DiskAPI for Disk { } } - async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result> { + async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await, Disk::Remote(remote_disk) => remote_disk.walk_dir(opts, wr).await, @@ -406,8 +406,8 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn stat_volume(&self, volume: &str) -> Result; async fn delete_volume(&self, volume: &str) -> Result<()>; - // 并发边读边写 TODO: wr io.Writer - async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result>; + // 并发边读边写 w <- MetaCacheEntry + async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()>; // Metadata operations async fn delete_version( diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 7a405a235..760b53927 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -351,7 +351,7 @@ impl DiskAPI for RemoteDisk { } // FIXME: TODO: use writer - async fn walk_dir(&self, opts: WalkDirOptions, _wr: &mut W) -> Result> { + async fn walk_dir(&self, opts: WalkDirOptions, _wr: &mut W) -> Result<()> { info!("walk_dir"); let walk_dir_options = serde_json::to_string(&opts)?; let mut client = node_service_time_out_client(&self.addr) @@ -362,19 +362,22 @@ impl DiskAPI for RemoteDisk { walk_dir_options, }); - let response = client.walk_dir(request).await?.into_inner(); + // TODO: use writer + unimplemented!() - if !response.success { - return Err(Error::from_string(response.error_info.unwrap_or("".to_string()))); - } + // let response = client.walk_dir(request).await?.into_inner(); - let entries = response - .meta_cache_entry - .into_iter() - .filter_map(|json_str| serde_json::from_str::(&json_str).ok()) - .collect(); + // if !response.success { + // return Err(Error::from_string(response.error_info.unwrap_or("".to_string()))); + // } - Ok(entries) + // let entries = response + // .meta_cache_entry + // .into_iter() + // .filter_map(|json_str| serde_json::from_str::(&json_str).ok()) + // .collect(); + + // Ok(entries) } async fn rename_data( diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 76ab932d8..cd170d907 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -1402,42 +1402,42 @@ impl SetDisks { Ok((disk, fm)) } - pub async fn walk_dir(&self, opts: &WalkDirOptions) -> (Vec>>, Vec>) { - let disks = self.disks.read().await; + // pub async fn walk_dir(&self, opts: &WalkDirOptions) -> (Vec>>, Vec>) { + // let disks = self.disks.read().await; - let disks = disks.clone(); - let mut futures = Vec::new(); - let mut errs = Vec::new(); - let mut ress = Vec::new(); + // let disks = disks.clone(); + // let mut futures = Vec::new(); + // let mut errs = Vec::new(); + // let mut ress = Vec::new(); - for disk in disks.iter() { - let opts = opts.clone(); - futures.push(async move { - if let Some(disk) = disk { - disk.walk_dir(opts, &mut Writer::NotUse).await - } else { - Err(Error::new(DiskError::DiskNotFound)) - } - }); - } + // for disk in disks.iter() { + // let opts = opts.clone(); + // futures.push(async move { + // if let Some(disk) = disk { + // disk.walk_dir(opts, &mut Writer::NotUse).await + // } else { + // Err(Error::new(DiskError::DiskNotFound)) + // } + // }); + // } - let results = join_all(futures).await; + // let results = join_all(futures).await; - for res in results { - match res { - Ok(entrys) => { - ress.push(Some(entrys)); - errs.push(None); - } - Err(e) => { - ress.push(None); - errs.push(Some(e)); - } - } - } + // for res in results { + // match res { + // Ok(entrys) => { + // ress.push(Some(entrys)); + // errs.push(None); + // } + // Err(e) => { + // ress.push(None); + // errs.push(Some(e)); + // } + // } + // } - (ress, errs) - } + // (ress, errs) + // } async fn remove_object_part( &self, diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index a2be08bac..d29ace0a4 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -731,43 +731,45 @@ impl Node for NodeService { } async fn walk_dir(&self, request: Request) -> Result, Status> { - let request = request.into_inner(); - if let Some(disk) = self.find_disk(&request.disk).await { - let opts = match serde_json::from_str::(&request.walk_dir_options) { - Ok(options) => options, - Err(_) => { - return Ok(tonic::Response::new(WalkDirResponse { - success: false, - meta_cache_entry: Vec::new(), - error_info: Some("can not decode DeleteOptions".to_string()), - })); - } - }; - match disk.walk_dir(opts, &mut ecstore::io::Writer::NotUse).await { - Ok(entries) => { - let entries = entries - .into_iter() - .filter_map(|entry| serde_json::to_string(&entry).ok()) - .collect(); - Ok(tonic::Response::new(WalkDirResponse { - success: true, - meta_cache_entry: entries, - error_info: None, - })) - } - Err(err) => Ok(tonic::Response::new(WalkDirResponse { - success: false, - meta_cache_entry: Vec::new(), - error_info: Some(err.to_string()), - })), - } - } else { - Ok(tonic::Response::new(WalkDirResponse { - success: false, - meta_cache_entry: Vec::new(), - error_info: Some("can not find disk".to_string()), - })) - } + // TODO: use writer + unimplemented!() + // let request = request.into_inner(); + // if let Some(disk) = self.find_disk(&request.disk).await { + // let opts = match serde_json::from_str::(&request.walk_dir_options) { + // Ok(options) => options, + // Err(_) => { + // return Ok(tonic::Response::new(WalkDirResponse { + // success: false, + // meta_cache_entry: Vec::new(), + // error_info: Some("can not decode DeleteOptions".to_string()), + // })); + // } + // }; + // match disk.walk_dir(opts, &mut ecstore::io::Writer::NotUse).await { + // Ok(entries) => { + // let entries = entries + // .into_iter() + // .filter_map(|entry| serde_json::to_string(&entry).ok()) + // .collect(); + // Ok(tonic::Response::new(WalkDirResponse { + // success: true, + // meta_cache_entry: entries, + // error_info: None, + // })) + // } + // Err(err) => Ok(tonic::Response::new(WalkDirResponse { + // success: false, + // meta_cache_entry: Vec::new(), + // error_info: Some(err.to_string()), + // })), + // } + // } else { + // Ok(tonic::Response::new(WalkDirResponse { + // success: false, + // meta_cache_entry: Vec::new(), + // error_info: Some("can not find disk".to_string()), + // })) + // } } async fn rename_data(&self, request: Request) -> Result, Status> {