gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 20:06:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

todo: rpc walk_dir use writer

Browse Source
This commit is contained in:
weisd
2024-12-22 10:52:52 +08:00
parent 5158ca14c9
commit 67ea60a0fa
5 changed files with 89 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/disk/local.rs
+2 -2
View File
@@ -1521,7 +1521,7 @@ impl DiskAPI for LocalDisk {
// FIXME: TODO: io.writer TODO cancel // FIXME: TODO: io.writer TODO cancel
#[tracing::instrument(level = "debug", skip(self, wr))] #[tracing::instrument(level = "debug", skip(self, wr))]
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<Vec<MetaCacheEntry>> { async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
let volume_dir = self.get_bucket_path(&opts.bucket)?; let volume_dir = self.get_bucket_path(&opts.bucket)?;
if !skip_access_checks(&opts.bucket) { if !skip_access_checks(&opts.bucket) {
@@ -1560,7 +1560,7 @@ impl DiskAPI for LocalDisk {
let mut current = opts.base_dir.clone(); let mut current = opts.base_dir.clone();
self.scan_dir(&mut current, &opts, &mut out, &mut objs_returned).await?; self.scan_dir(&mut current, &opts, &mut out, &mut objs_returned).await?;
Ok(Vec::new()) Ok(())
} }
// #[tracing::instrument(skip(self))] // #[tracing::instrument(skip(self))]
ecstore/src/disk/mod.rs
+3 -3
View File
@@ -211,7 +211,7 @@ impl DiskAPI for Disk {
} }
} }
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<Vec<MetaCacheEntry>> { async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
match self { match self {
Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await, Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await,
Disk::Remote(remote_disk) => remote_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<VolumeInfo>; async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo>;
async fn delete_volume(&self, volume: &str) -> Result<()>; async fn delete_volume(&self, volume: &str) -> Result<()>;
// 并发边读边写 TODO: wr io.Writer // 并发边读边写 w <- MetaCacheEntry
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<Vec<MetaCacheEntry>>; async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()>;
// Metadata operations // Metadata operations
async fn delete_version( async fn delete_version(
ecstore/src/disk/remote.rs
+14 -11
View File
@@ -351,7 +351,7 @@ impl DiskAPI for RemoteDisk {
} }
// FIXME: TODO: use writer // FIXME: TODO: use writer
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, _wr: &mut W) -> Result<Vec<MetaCacheEntry>> { async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, _wr: &mut W) -> Result<()> {
info!("walk_dir"); info!("walk_dir");
let walk_dir_options = serde_json::to_string(&opts)?; let walk_dir_options = serde_json::to_string(&opts)?;
let mut client = node_service_time_out_client(&self.addr) let mut client = node_service_time_out_client(&self.addr)
@@ -362,19 +362,22 @@ impl DiskAPI for RemoteDisk {
walk_dir_options, walk_dir_options,
}); });
let response = client.walk_dir(request).await?.into_inner(); // TODO: use writer
unimplemented!()
if !response.success { // let response = client.walk_dir(request).await?.into_inner();
return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
}
let entries = response // if !response.success {
.meta_cache_entry // return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
.into_iter() // }
.filter_map(|json_str| serde_json::from_str::<MetaCacheEntry>(&json_str).ok())
.collect();
Ok(entries) // let entries = response
// .meta_cache_entry
// .into_iter()
// .filter_map(|json_str| serde_json::from_str::<MetaCacheEntry>(&json_str).ok())
// .collect();
// Ok(entries)
} }
async fn rename_data( async fn rename_data(
ecstore/src/set_disk.rs
+31 -31
View File
@@ -1402,42 +1402,42 @@ impl SetDisks {
Ok((disk, fm)) Ok((disk, fm))
} }
pub async fn walk_dir(&self, opts: &WalkDirOptions) -> (Vec<Option<Vec<MetaCacheEntry>>>, Vec<Option<Error>>) { // pub async fn walk_dir(&self, opts: &WalkDirOptions) -> (Vec<Option<Vec<MetaCacheEntry>>>, Vec<Option<Error>>) {
let disks = self.disks.read().await; // let disks = self.disks.read().await;
let disks = disks.clone(); // let disks = disks.clone();
let mut futures = Vec::new(); // let mut futures = Vec::new();
let mut errs = Vec::new(); // let mut errs = Vec::new();
let mut ress = Vec::new(); // let mut ress = Vec::new();
for disk in disks.iter() { // for disk in disks.iter() {
let opts = opts.clone(); // let opts = opts.clone();
futures.push(async move { // futures.push(async move {
if let Some(disk) = disk { // if let Some(disk) = disk {
disk.walk_dir(opts, &mut Writer::NotUse).await // disk.walk_dir(opts, &mut Writer::NotUse).await
} else { // } else {
Err(Error::new(DiskError::DiskNotFound)) // Err(Error::new(DiskError::DiskNotFound))
} // }
}); // });
} // }
let results = join_all(futures).await; // let results = join_all(futures).await;
for res in results { // for res in results {
match res { // match res {
Ok(entrys) => { // Ok(entrys) => {
ress.push(Some(entrys)); // ress.push(Some(entrys));
errs.push(None); // errs.push(None);
} // }
Err(e) => { // Err(e) => {
ress.push(None); // ress.push(None);
errs.push(Some(e)); // errs.push(Some(e));
} // }
} // }
} // }
(ress, errs) // (ress, errs)
} // }
async fn remove_object_part( async fn remove_object_part(
&self, &self,
rustfs/src/grpc.rs
+39 -37
View File
@@ -731,43 +731,45 @@ impl Node for NodeService {
} }
async fn walk_dir(&self, request: Request<WalkDirRequest>) -> Result<Response<WalkDirResponse>, Status> { async fn walk_dir(&self, request: Request<WalkDirRequest>) -> Result<Response<WalkDirResponse>, Status> {
let request = request.into_inner(); // TODO: use writer
if let Some(disk) = self.find_disk(&request.disk).await { unimplemented!()
let opts = match serde_json::from_str::<WalkDirOptions>(&request.walk_dir_options) { // let request = request.into_inner();
Ok(options) => options, // if let Some(disk) = self.find_disk(&request.disk).await {
Err(_) => { // let opts = match serde_json::from_str::<WalkDirOptions>(&request.walk_dir_options) {
return Ok(tonic::Response::new(WalkDirResponse { // Ok(options) => options,
success: false, // Err(_) => {
meta_cache_entry: Vec::new(), // return Ok(tonic::Response::new(WalkDirResponse {
error_info: Some("can not decode DeleteOptions".to_string()), // 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 // match disk.walk_dir(opts, &mut ecstore::io::Writer::NotUse).await {
.into_iter() // Ok(entries) => {
.filter_map(|entry| serde_json::to_string(&entry).ok()) // let entries = entries
.collect(); // .into_iter()
Ok(tonic::Response::new(WalkDirResponse { // .filter_map(|entry| serde_json::to_string(&entry).ok())
success: true, // .collect();
meta_cache_entry: entries, // Ok(tonic::Response::new(WalkDirResponse {
error_info: None, // success: true,
})) // meta_cache_entry: entries,
} // error_info: None,
Err(err) => Ok(tonic::Response::new(WalkDirResponse { // }))
success: false, // }
meta_cache_entry: Vec::new(), // Err(err) => Ok(tonic::Response::new(WalkDirResponse {
error_info: Some(err.to_string()), // success: false,
})), // meta_cache_entry: Vec::new(),
} // error_info: Some(err.to_string()),
} else { // })),
Ok(tonic::Response::new(WalkDirResponse { // }
success: false, // } else {
meta_cache_entry: Vec::new(), // Ok(tonic::Response::new(WalkDirResponse {
error_info: Some("can not find disk".to_string()), // success: false,
})) // meta_cache_entry: Vec::new(),
} // error_info: Some("can not find disk".to_string()),
// }))
// }
} }
async fn rename_data(&self, request: Request<RenameDataRequest>) -> Result<Response<RenameDataResponse>, Status> { async fn rename_data(&self, request: Request<RenameDataRequest>) -> Result<Response<RenameDataResponse>, Status> {