list_object v2 use channel, todo other params

ecstore/src/disk/error.rs

@@ -452,6 +452,10 @@ pub fn is_all_not_found(errs: &[Option<Error>]) -> bool {
     !errs.is_empty()
 }
 
+pub fn is_all_volume_not_found(errs: &[Option<Error>]) -> bool {
+    DiskError::VolumeNotFound.count_errs(errs) == errs.len()
+}
+
 pub fn is_all_buckets_not_found(errs: &[Option<Error>]) -> bool {
     if errs.is_empty() {
         return false;

ecstore/src/disk/local.rs

@@ -439,7 +439,11 @@ impl LocalDisk {
                             }
                         }
 
-                        Err(err)
+                        if let Some(os_err) = err.downcast_ref::<std::io::Error>() {
+                            Err(os_err_to_file_err(std::io::Error::new(os_err.kind(), os_err.to_string())))
+                        } else {
+                            Err(err)
+                        }
                     }
                 }
             }
@@ -531,7 +535,7 @@ impl LocalDisk {
                     return Err(Error::new(DiskError::UnsupportedDisk));
                 }
 
-                return Err(Error::new(e));
+                return Err(os_err_to_file_err(e));
             }
         };
 
@@ -848,7 +852,7 @@ impl LocalDisk {
                 continue;
             }
 
-            let name = path::path_join_buf(&[current, &entry]);
+            let name = path::path_join_buf(&[current, entry]);
 
             if !dir_stack.is_empty() {
                 if let Some(pop) = dir_stack.pop() {
@@ -1514,6 +1518,7 @@ impl DiskAPI for LocalDisk {
                 return Err(e);
             }
         };
+
         Ok(entries)
     }
 

ecstore/src/disk/mod.rs

@@ -15,6 +15,10 @@ pub const STORAGE_FORMAT_FILE: &str = "xl.meta";
 pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp";
 
 use crate::{
+    bucket::{
+        metadata_sys::get_versioning_config,
+        versioning::{self, VersioningApi},
+    },
     erasure::Writer,
     error::{Error, Result},
     file_meta::{merge_file_meta_versions, FileMeta, FileMetaShallowVersion},
@@ -24,7 +28,7 @@ use crate::{
         heal_commands::{HealScanMode, HealingTracker},
     },
     io,
-    store_api::{FileInfo, RawFileInfo},
+    store_api::{FileInfo, ObjectInfo, RawFileInfo},
 };
 use endpoint::Endpoint;
 use error::DiskError;
@@ -41,6 +45,7 @@ use std::{
     cmp::Ordering,
     fmt::Debug,
     io::{Cursor, SeekFrom},
+    ops::Index,
     path::PathBuf,
     sync::Arc,
 };
@@ -795,6 +800,10 @@ impl MetaCacheEntry {
 pub struct MetaCacheEntries(pub Vec<Option<MetaCacheEntry>>);
 
 impl MetaCacheEntries {
+    #[allow(clippy::should_implement_trait)]
+    pub fn as_ref(&self) -> &[Option<MetaCacheEntry>] {
+        &self.0
+    }
     pub fn resolve(&self, mut params: MetadataResolutionParams) -> Result<Option<MetaCacheEntry>> {
         if self.0.is_empty() {
             return Ok(None);
@@ -885,6 +894,83 @@ impl MetaCacheEntries {
     }
 }
 
+#[derive(Debug)]
+pub struct MetaCacheEntriesSorted {
+    pub o: MetaCacheEntries,
+    // pub list_id: String,
+    // pub reuse: bool,
+    // pub lastSkippedEntry: String,
+}
+
+impl MetaCacheEntriesSorted {
+    pub fn entries(&self) -> Vec<MetaCacheEntry> {
+        let entries: Vec<MetaCacheEntry> = self.o.0.iter().flatten().cloned().collect();
+        entries
+    }
+    pub async fn file_infos(&self, bucket: &str, prefix: &str, delimiter: &str) -> Vec<ObjectInfo> {
+        let vcfg = get_versioning_config(bucket).await.ok();
+        let mut objects = Vec::with_capacity(self.o.as_ref().len());
+        let mut prev_prefix = "";
+        for entry in self.o.as_ref().iter().flatten() {
+            if entry.is_object() {
+                if !delimiter.is_empty() {
+                    if let Some(idx) = entry.name.trim_start_matches(prefix).find(delimiter) {
+                        let idx = prefix.len() + idx + delimiter.len();
+                        if let Some(curr_prefix) = entry.name.get(0..idx) {
+                            if curr_prefix == prev_prefix {
+                                continue;
+                            }
+
+                            prev_prefix = curr_prefix.clone();
+
+                            objects.push(ObjectInfo {
+                                is_dir: true,
+                                bucket: bucket.to_owned(),
+                                name: curr_prefix.to_owned(),
+                                ..Default::default()
+                            });
+                        }
+                        continue;
+                    }
+                }
+
+                if let Ok(Some(fi)) = entry.to_fileinfo(bucket) {
+                    // TODO:VersionPurgeStatus
+                    let versioned = vcfg.clone().map(|v| v.0.versioned(&entry.name)).unwrap_or_default();
+                    objects.push(fi.to_object_info(bucket, &entry.name, versioned));
+                }
+                continue;
+            }
+
+            if entry.is_dir() {
+                if delimiter.is_empty() {
+                    continue;
+                }
+
+                if let Some(idx) = entry.name.trim_start_matches(prefix).find(delimiter) {
+                    let idx = prefix.len() + idx + delimiter.len();
+                    if let Some(curr_prefix) = entry.name.get(0..idx) {
+                        if curr_prefix == prev_prefix {
+                            continue;
+                        }
+
+                        prev_prefix = curr_prefix.clone();
+
+                        objects.push(ObjectInfo {
+                            is_dir: true,
+                            bucket: bucket.to_owned(),
+                            name: curr_prefix.to_owned(),
+                            ..Default::default()
+                        });
+                    }
+                }
+            }
+        }
+
+        objects
+    }
+}
+
 #[derive(Clone, Debug, Default)]
 pub struct DiskOption {
     pub cleanup: bool,

ecstore/src/metacache/writer.rs

@@ -7,6 +7,7 @@ use tokio::io::AsyncRead;
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
+use tracing::warn;
 // use std::sync::Arc;
 // use tokio::sync::mpsc;
 // use tokio::sync::mpsc::Sender;

ecstore/src/set_disk.rs

@@ -164,12 +164,13 @@ impl SetDisks {
         let mut infos = Vec::with_capacity(disks.len());
 
         let mut futures = Vec::with_capacity(disks.len());
-
-        let mut rng = thread_rng();
-        disks.shuffle(&mut rng);
-
         let mut numbers: Vec<usize> = (0..disks.len()).collect();
-        numbers.shuffle(&mut rand::thread_rng());
+        {
+            let mut rng = thread_rng();
+            disks.shuffle(&mut rng);
+
+            numbers.shuffle(&mut rand::thread_rng());
+        }
 
         for &i in numbers.iter() {
             let disk = disks[i].clone();
@@ -818,6 +819,7 @@ impl SetDisks {
         };
 
         if let Some(err) = reduce_read_quorum_errs(errs, object_op_ignored_errs().as_ref(), expected_rquorum) {
+            // warn!("object_quorum_from_meta err {:?}", &err);
             return Err(err);
         }
 
@@ -1687,11 +1689,12 @@ impl SetDisks {
         // TODO: 优化并发 可用数量中断
         let (parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false).await;
         // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
-        warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);
+        // warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);
 
         let _min_disks = self.set_drive_count - self.default_parity_count;
 
-        let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?;
+        let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)
+            .map_err(|err| to_object_err(err, vec![bucket, object]))?;
 
         if let Some(err) = reduce_read_quorum_errs(&errs, object_op_ignored_errs().as_ref(), read_quorum as usize) {
             error!(
@@ -1700,7 +1703,7 @@ impl SetDisks {
                 read_quorum,
                 &errs
             );
-            return Err(err);
+            return Err(to_object_err(err, vec![bucket, object]));
         }
 
         let (op_online_disks, mot_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum as usize);
@@ -3832,7 +3835,7 @@ impl StorageAPI for SetDisks {
     }
 
     async fn list_objects_v2(
-        &self,
+        self: Arc<Self>,
         _bucket: &str,
         _prefix: &str,
         _continuation_token: &str,

ecstore/src/sets.rs

@@ -451,7 +451,7 @@ impl StorageAPI for Sets {
         self.get_disks_by_key(object).delete_object(bucket, object, opts).await
     }
     async fn list_objects_v2(
-        &self,
+        self: Arc<Self>,
         _bucket: &str,
         _prefix: &str,
         _continuation_token: &str,

ecstore/src/store.rs

@@ -1512,7 +1512,7 @@ impl StorageAPI for ECStore {
 
     // TODO: review
     async fn list_objects_v2(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         prefix: &str,
         continuation_token: &str,
@@ -2170,9 +2170,11 @@ async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, po
     *GLOBAL_Local_Node_Name.write().await = peer_set[0].clone();
 }
 
-pub fn is_valid_object_prefix(object: &str) -> bool {
+pub fn is_valid_object_prefix(_object: &str) -> bool {
     // Implement object prefix validation
-    !object.is_empty() // Placeholder
+    // !object.is_empty() // Placeholder
+    // FIXME: TODO:
+    true
 }
 
 fn is_valid_object_name(object: &str) -> bool {

ecstore/src/store_api.rs

@@ -843,7 +843,7 @@ pub trait StorageAPI: ObjectIO {
     async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
     // ListObjects TODO: FIXME:
     async fn list_objects_v2(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         prefix: &str,
         continuation_token: &str,

ecstore/src/store_list_objects.rs

@@ -1,5 +1,8 @@
 use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions};
-use crate::disk::{DiskInfo, DiskStore, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, WalkDirOptions};
+use crate::disk::error::{is_all_not_found, is_all_volume_not_found, is_err_eof, DiskError};
+use crate::disk::{
+    DiskInfo, DiskStore, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntry, MetadataResolutionParams, WalkDirOptions,
+};
 use crate::error::{Error, Result};
 use crate::file_meta::merge_file_meta_versions;
 use crate::peer::is_reserved_or_invalid_bucket;
@@ -14,9 +17,10 @@ use rand::seq::SliceRandom;
 use rand::thread_rng;
 use std::collections::{HashMap, HashSet};
 use std::io::ErrorKind;
-use tokio::sync::broadcast::Receiver as B_Receiver;
+use std::sync::Arc;
+use tokio::sync::broadcast::{self, Receiver as B_Receiver};
 use tokio::sync::mpsc::{self, Receiver, Sender};
-use tracing::error;
+use tracing::{error, warn};
 
 const MAX_OBJECT_LIST: i32 = 1000;
 const MAX_DELETE_LIST: i32 = 1000;
@@ -114,7 +118,7 @@ impl ListPathOptions {
 impl ECStore {
     #[allow(clippy::too_many_arguments)]
     pub async fn inner_list_objects_v2(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         prefix: &str,
         continuation_token: &str,
@@ -131,13 +135,18 @@ impl ECStore {
             }
         };
 
-        self.list_objects_generic(bucket, prefix, marker, delimiter, max_keys).await?;
-        // FIXME:TODO:
-        unimplemented!()
+        let loi = self.list_objects_generic(bucket, prefix, marker, delimiter, max_keys).await?;
+        Ok(ListObjectsV2Info {
+            is_truncated: loi.is_truncated,
+            continuation_token: continuation_token.to_owned(),
+            next_continuation_token: loi.next_marker,
+            objects: loi.objects,
+            prefixes: loi.prefixes,
+        })
     }
 
     pub async fn list_objects_generic(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         prefix: &str,
         marker: &str,
@@ -155,14 +164,73 @@ impl ECStore {
             ..Default::default()
         };
 
-        let merged = self.list_path(&opts).await?;
+        let mut err_eof = false;
+        let has_merged = match self.list_path(&opts).await {
+            Ok(res) => Some(res),
+            Err(err) => {
+                if !is_err_eof(&err) {
+                    return Err(err);
+                }
 
-        // FIXME:TODO:
+                err_eof = true;
+                None
+            }
+        };
 
-        todo!()
+        let mut get_objects = if let Some(merged) = has_merged {
+            merged.file_infos(bucket, prefix, delimiter).await
+        } else {
+            Vec::new()
+        };
+
+        let is_truncated = {
+            if max_keys > 0 && get_objects.len() > max_keys as usize {
+                get_objects.truncate(max_keys as usize);
+                true
+            } else {
+                !err_eof && get_objects.len() > 0
+            }
+        };
+
+        let mut prefixes: Vec<String> = Vec::new();
+
+        let mut objects = Vec::with_capacity(get_objects.len());
+        for obj in get_objects.into_iter() {
+            if obj.is_dir && obj.mod_time.is_none() && !delimiter.is_empty() {
+                let mut found = false;
+                if delimiter != SLASH_SEPARATOR {
+                    for p in prefixes.iter() {
+                        if found {
+                            break;
+                        }
+                        found = p == &obj.name;
+                    }
+                }
+                if !found {
+                    prefixes.push(obj.name.clone());
+                }
+            } else {
+                objects.push(obj);
+            }
+        }
+
+        let next_marker = {
+            if is_truncated {
+                objects.last().map(|last| last.name.clone())
+            } else {
+                None
+            }
+        };
+
+        Ok(ListObjectsInfo {
+            is_truncated,
+            next_marker: next_marker.unwrap_or_default(),
+            objects,
+            prefixes,
+        })
     }
 
-    pub async fn list_path(&self, o: &ListPathOptions) -> Result<ListObjectsInfo> {
+    pub async fn list_path(self: Arc<Self>, o: &ListPathOptions) -> Result<MetaCacheEntriesSorted> {
         check_list_objs_args(&o.bucket, &o.prefix, &o.marker)?;
         // if opts.prefix.ends_with(SLASH_SEPARATOR) {
         //     return Err(Error::msg("eof"));
@@ -181,7 +249,7 @@ impl ECStore {
             return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
         }
 
-        if o.prefix.ends_with(SLASH_SEPARATOR) {
+        if o.prefix.starts_with(SLASH_SEPARATOR) {
             return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
         }
 
@@ -207,7 +275,29 @@ impl ECStore {
         }
 
         // FIXME:TODO:
-        todo!()
+        let (tx, rx) = broadcast::channel(1);
+
+        let (sender, mut recv) = mpsc::channel(o.limit as usize);
+
+        let store = self.clone();
+        let opts = o.clone();
+        let rx1 = rx.resubscribe();
+        tokio::spawn(async move {
+            if let Err(err) = store.list_merged(rx1, opts, sender).await {
+                error!("list_merged err {:?}", err);
+            }
+        });
+
+        let rx2 = rx.resubscribe();
+        let res = gather_results(rx2, &o, &mut recv).await?;
+
+        tx.send(true)?;
+
+        // TODO: recv list_merged err
+
+        Ok(MetaCacheEntriesSorted {
+            o: MetaCacheEntries(res.into_iter().map(Some).collect()),
+        })
 
         // let mut opts = opts.clone();
 
@@ -247,20 +337,57 @@ impl ECStore {
             }
         }
 
-        let merge_res = merge_entry_channels(rx, inputs, sender.clone(), 1).await;
+        tokio::spawn(async move {
+            if let Err(err) = merge_entry_channels(rx, inputs, sender.clone(), 1).await {
+                println!("merge_entry_channels err {:?}", err)
+            }
+        });
+
+        // let merge_res = merge_entry_channels(rx, inputs, sender.clone(), 1).await;
 
         let results = join_all(futures).await;
 
+        let mut all_at_eof = true;
+
         let mut errs = Vec::new();
         for result in results {
             if let Err(err) = result {
+                all_at_eof = false;
                 errs.push(Some(err));
             } else {
                 errs.push(None);
             }
         }
 
-        unimplemented!()
+        if is_all_not_found(&errs) {
+            if is_all_volume_not_found(&errs) {
+                return Err(Error::new(DiskError::VolumeNotFound));
+            }
+
+            return Ok(Vec::new());
+        }
+
+        // merge_res?;
+
+        // TODO check cancel
+
+        for err in errs.iter() {
+            if let Some(err) = err {
+                if is_err_eof(err) {
+                    continue;
+                }
+
+                return Err(err.clone());
+            } else {
+                all_at_eof = false;
+                continue;
+            }
+        }
+
+        // check all_at_eof
+        _ = all_at_eof;
+
+        Ok(Vec::new())
 
         // // let mut errs = Vec::new();
         // let mut ress = Vec::new();
@@ -320,6 +447,42 @@ impl ECStore {
     }
 }
 
+// TODO: FIXME: 异步
+async fn gather_results(
+    _rx: B_Receiver<bool>,
+    opts: &ListPathOptions,
+    recv: &mut Receiver<MetaCacheEntry>,
+) -> Result<Vec<MetaCacheEntry>> {
+    let mut returned = false;
+    let mut results = Vec::new();
+    while let Some(entry) = recv.recv().await {
+        if returned {
+            continue;
+        }
+
+        // TODO: rx.recv()
+
+        // TODO: isLatestDeletemarker
+        if !opts.include_directories && (entry.is_dir() || (!opts.versioned && entry.is_object())) {
+            continue;
+        }
+
+        if !opts.marker.is_empty() && entry.name < opts.marker {
+            continue;
+        }
+
+        // TODO: other
+        if opts.limit > 0 && results.len() >= opts.limit as usize {
+            returned = true;
+            continue;
+        }
+
+        results.push(entry);
+    }
+
+    Ok(results)
+}
+
 async fn select_from(
     in_channels: &mut [Receiver<MetaCacheEntry>],
     idx: usize,
@@ -390,6 +553,8 @@ async fn merge_entry_channels(
                 if let Some(best_entry) = &best {
                     let other_idx = i;
 
+                    // println!("get other_entry {:?}", other_entry.name);
+
                     if path::clean(&best_entry.name) == path::clean(&other_entry.name) {
                         let dir_matches = best_entry.is_dir() && other_entry.is_dir();
                         let suffix_matche =
@@ -426,6 +591,8 @@ async fn merge_entry_channels(
             }
         }
 
+        // println!("get best_entry {} {:?}", &best_idx, &best.clone().unwrap_or_default().name);
+
         // TODO:
         if !to_merge.is_empty() {
             if let Some(entry) = &best {
@@ -667,10 +834,13 @@ mod test {
     use crate::disk::MetaCacheEntries;
     use crate::disk::MetaCacheEntry;
     use crate::disk::WalkDirOptions;
+    use crate::endpoints::EndpointServerPools;
     use crate::error::Error;
     use crate::metacache::writer::MetacacheReader;
     use crate::set_disk::SetDisks;
+    use crate::store::ECStore;
     use crate::store_list_objects::ListPathOptions;
+    use crate::StorageAPI;
     use futures::future::join_all;
     use lock::namespace_lock::NsLockMap;
     use tokio::sync::broadcast;
@@ -788,7 +958,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_list_path() {
+    async fn test_set_list_path() {
         let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap();
         ep.pool_idx = 0;
         ep.set_idx = 0;
@@ -829,4 +999,81 @@ mod test {
             println!("get entry {:?}", entry.name)
         }
     }
+
+    #[tokio::test]
+    async fn test_list_merged() {
+        let server_address = "localhost:9000";
+
+        let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
+            server_address,
+            vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
+        )
+        .unwrap();
+
+        let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
+            .await
+            .unwrap();
+
+        let (_tx, rx) = broadcast::channel(1);
+
+        let bucket = "dada".to_owned();
+        let opts = ListPathOptions {
+            bucket,
+            recursive: true,
+            ..Default::default()
+        };
+
+        let (sender, mut recv) = mpsc::channel(10);
+
+        store.list_merged(rx, opts, sender).await.unwrap();
+
+        while let Some(entry) = recv.recv().await {
+            println!("get entry {:?}", entry.name)
+        }
+    }
+
+    #[tokio::test]
+    async fn test_list_path() {
+        let server_address = "localhost:9000";
+
+        let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
+            server_address,
+            vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
+        )
+        .unwrap();
+
+        let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
+            .await
+            .unwrap();
+
+        let bucket = "dada".to_owned();
+        let opts = ListPathOptions {
+            bucket,
+            recursive: true,
+            limit: 100,
+
+            ..Default::default()
+        };
+
+        let ret = store.list_path(&opts).await.unwrap();
+        println!("ret {:?}", ret);
+    }
+
+    // #[tokio::test]
+    // async fn test_list_objects_v2() {
+    //     let server_address = "localhost:9000";
+
+    //     let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
+    //         server_address,
+    //         vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
+    //     )
+    //     .unwrap();
+
+    //     let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
+    //         .await
+    //         .unwrap();
+
+    //     let ret = store.list_objects_v2("data", "", "", "", 100, false, "").await.unwrap();
+    //     println!("ret {:?}", ret);
+    // }
 }