listobject v1 未使用chan

ecstore/src/disk/local.rs

@@ -12,15 +12,13 @@ use crate::{
 };
 use bytes::Bytes;
 use path_absolutize::Absolutize;
-use std::sync::Arc;
 use std::{
     fs::Metadata,
     path::{Path, PathBuf},
 };
 use time::OffsetDateTime;
 use tokio::fs::{self, File};
-use tokio::io::{DuplexStream, ErrorKind};
-use tokio::sync::mpsc;
+use tokio::io::ErrorKind;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
@@ -526,7 +524,8 @@ impl DiskAPI for LocalDisk {
 
         Ok(volumes)
     }
-    async fn walk_dir(&self, opts: WalkDirOptions, wr: Arc<DuplexStream>) -> Result<()> {
+
+    async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
         let mut entries = self.list_dir("", &opts.bucket, &opts.base_dir, -1).await?;
 
         entries.sort();
@@ -536,11 +535,13 @@ impl DiskAPI for LocalDisk {
 
         let bucket = opts.bucket.as_str();
 
+        let mut metas = Vec::new();
+
         // 第一层过滤
         for entry in entries.iter() {
             // check limit
             if opts.limit > 0 && objs_returned >= opts.limit {
-                return Ok(());
+                return Ok(metas);
             }
             // check prefix
             if !opts.filter_prefix.is_empty() && !entry.starts_with(&opts.filter_prefix) {
@@ -558,7 +559,7 @@ impl DiskAPI for LocalDisk {
 
             let (fdata, _) = match self.read_metadata_with_dmtime(&fpath).await {
                 Ok(res) => res,
-                Err(e) => {
+                Err(_) => {
                     // TODO: check err
                     (Vec::new(), OffsetDateTime::UNIX_EPOCH)
                 }
@@ -566,10 +567,10 @@ impl DiskAPI for LocalDisk {
 
             meta.metadata = fdata;
 
-            // TODO: FIXME:
+            metas.push(meta);
         }
 
-        Ok(())
+        Ok(metas)
     }
 
     // #[tracing::instrument(skip(self))]

ecstore/src/disk/mod.rs

@@ -22,7 +22,7 @@ use std::{fmt::Debug, io::SeekFrom, pin::Pin, sync::Arc};
 use time::OffsetDateTime;
 use tokio::{
     fs::File,
-    io::{AsyncReadExt, AsyncSeekExt, AsyncWrite, DuplexStream},
+    io::{AsyncReadExt, AsyncSeekExt, AsyncWrite},
 };
 use uuid::Uuid;
 
@@ -53,7 +53,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     // 读目录下的所有文件、目录
     async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>>;
     // 并发边读边写 TODO: wr io.Writer
-    async fn walk_dir(&self, opts: WalkDirOptions, wr: Arc<DuplexStream>) -> Result<()>;
+    async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>>;
     async fn rename_data(
         &self,
         src_volume: &str,
@@ -112,16 +112,71 @@ pub struct WalkDirOptions {
 #[derive(Debug, Default)]
 pub struct MetaCacheEntry {
     // name is the full name of the object including prefixes
-    name: String,
+    pub name: String,
     // Metadata. If none is present it is not an object but only a prefix.
     // Entries without metadata will only be present in non-recursive scans.
-    metadata: Vec<u8>,
+    pub metadata: Vec<u8>,
 
     // cached contains the metadata if decoded.
     cached: Option<FileMeta>,
 
     // Indicates the entry can be reused and only one reference to metadata is expected.
-    reusable: bool,
+    _reusable: bool,
+}
+
+impl MetaCacheEntry {
+    pub fn marshal_msg(&self) -> Result<Vec<u8>> {
+        let mut wr = Vec::new();
+        rmp::encode::write_bool(&mut wr, true)?;
+
+        rmp::encode::write_str(&mut wr, &self.name)?;
+
+        rmp::encode::write_bin(&mut wr, &self.metadata)?;
+
+        Ok(wr)
+    }
+    pub fn is_dir(&self) -> bool {
+        self.metadata.is_empty() && self.name.ends_with("/")
+    }
+    pub fn is_object(&self) -> bool {
+        !self.metadata.is_empty()
+    }
+
+    #[tracing::instrument(level = "debug", skip(self))]
+    pub fn to_fileinfo(&self, bucket: &str) -> Result<Option<FileInfo>> {
+        if self.is_dir() {
+            return Ok(Some(FileInfo {
+                volume: bucket.to_owned(),
+                name: self.name.clone(),
+                ..Default::default()
+            }));
+        }
+
+        if self.cached.is_some() {
+            let fm = self.cached.as_ref().unwrap();
+            if fm.versions.is_empty() {
+                return Ok(Some(FileInfo {
+                    volume: bucket.to_owned(),
+                    name: self.name.clone(),
+                    deleted: true,
+                    is_latest: true,
+                    mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+                    ..Default::default()
+                }));
+            }
+
+            let fi = fm.into_fileinfo(bucket, self.name.as_str(), "", false, false)?;
+
+            return Ok(Some(fi));
+        }
+
+        let mut fm = FileMeta::new();
+        fm.unmarshal_msg(&self.metadata)?;
+
+        let fi = fm.into_fileinfo(bucket, self.name.as_str(), "", false, false)?;
+
+        return Ok(Some(fi));
+    }
 }
 
 pub struct DiskOption {

ecstore/src/sets.rs

@@ -10,8 +10,8 @@ use crate::{
     error::Result,
     set_disk::SetDisks,
     store_api::{
-        BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsInfo, ListObjectsV2Info,
-        MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
+        BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsV2Info, MakeBucketOptions,
+        MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
     },
     utils::hash,
 };
@@ -137,13 +137,13 @@ impl StorageAPI for Sets {
 
     async fn list_objects_v2(
         &self,
-        bucket: &str,
-        prefix: &str,
-        continuation_token: &str,
-        delimiter: &str,
-        max_keys: i32,
-        fetch_owner: bool,
-        start_after: &str,
+        _bucket: &str,
+        _prefix: &str,
+        _continuation_token: &str,
+        _delimiter: &str,
+        _max_keys: i32,
+        _fetch_owner: bool,
+        _start_after: &str,
     ) -> Result<ListObjectsV2Info> {
         unimplemented!()
     }

ecstore/src/store.rs

@@ -1,6 +1,6 @@
 use crate::{
     bucket_meta::BucketMetadata,
-    disk::{error::DiskError, DiskOption, DiskStore, MetaCacheEntry, WalkDirOptions, RUSTFS_META_BUCKET},
+    disk::{error::DiskError, DiskOption, DiskStore, WalkDirOptions, RUSTFS_META_BUCKET},
     disks_layout::DisksLayout,
     endpoints::EndpointServerPools,
     error::{Error, Result},
@@ -15,8 +15,8 @@ use crate::{
 use futures::future::join_all;
 use http::HeaderMap;
 use s3s::{dto::StreamingBlob, Body};
-use std::{collections::HashMap, sync::Arc};
-use tokio::sync::mpsc;
+use std::collections::{HashMap, HashSet};
+use tracing::warn;
 use uuid::Uuid;
 
 #[derive(Debug)]
@@ -108,22 +108,24 @@ impl ECStore {
         self.pools.len() == 1
     }
 
-    async fn list_path(&self, opts: &ListPathOptions) -> Result<()> {
-        let (sender, rec) = mpsc::channel(64);
-        self.listMerged(opts, sender).await?;
-        Ok(())
+    async fn list_path(&self, opts: &ListPathOptions) -> Result<ListObjectsInfo> {
+        let objects = self.list_merged(opts).await?;
+
+        let info = ListObjectsInfo {
+            objects,
+            ..Default::default()
+        };
+        Ok(info)
     }
 
     // 读所有
-    async fn listMerged(&self, opts: &ListPathOptions, results: mpsc::Sender<MetaCacheEntry>) -> Result<()> {
+    async fn list_merged(&self, opts: &ListPathOptions) -> Result<Vec<ObjectInfo>> {
         let opts = WalkDirOptions {
             bucket: opts.bucket.clone(),
             ..Default::default()
         };
 
-        let (mut wr, mut rd) = tokio::io::duplex(1024);
-
-        let wr = Arc::new(wr);
+        // let (mut wr, mut rd) = tokio::io::duplex(1024);
 
         let mut futures = Vec::new();
 
@@ -136,8 +138,8 @@ impl ECStore {
 
                     let disk = disk.as_ref().unwrap();
                     let opts = opts.clone();
-                    let wr = wr.clone();
-                    futures.push(disk.walk_dir(opts, wr));
+                    // let mut wr = &mut wr;
+                    futures.push(disk.walk_dir(opts));
                     // tokio::spawn(async move { disk.walk_dir(opts, wr).await });
                 }
             }
@@ -146,20 +148,52 @@ impl ECStore {
         let results = join_all(futures).await;
 
         let mut errs = Vec::new();
+        let mut ress = Vec::new();
+        let mut uniq = HashSet::new();
 
         for res in results {
             match res {
-                Ok(_) => errs.push(None),
+                Ok(entrys) => {
+                    for entry in entrys {
+                        if !uniq.contains(&entry.name) {
+                            uniq.insert(entry.name.clone());
+                            // TODO: 过滤
+                            if opts.limit > 0 && ress.len() as i32 >= opts.limit {
+                                return Ok(ress);
+                            }
+
+                            if entry.is_object() {
+                                let fi = entry.to_fileinfo(&opts.bucket)?;
+                                if fi.is_some() {
+                                    ress.push(fi.unwrap().into_object_info(&opts.bucket, &entry.name, false));
+                                }
+                                continue;
+                            }
+
+                            if entry.is_dir() {
+                                ress.push(ObjectInfo {
+                                    is_dir: true,
+                                    bucket: opts.bucket.clone(),
+                                    name: entry.name,
+                                    ..Default::default()
+                                });
+                            }
+                        }
+                    }
+                    errs.push(None);
+                }
                 Err(e) => errs.push(Some(e)),
             }
         }
 
-        Ok(())
+        warn!("list_merged errs {:?}", errs);
+
+        Ok(ress)
     }
 }
 
 #[derive(Debug, Default)]
-struct ListPathOptions {
+pub struct ListPathOptions {
     pub id: String,
 
     // Bucket of the listing.
@@ -234,12 +268,12 @@ impl StorageAPI for ECStore {
     async fn list_objects_v2(
         &self,
         bucket: &str,
-        prefix: &str,
+        _prefix: &str,
         continuation_token: &str,
-        delimiter: &str,
+        _delimiter: &str,
         max_keys: i32,
-        fetch_owner: bool,
-        start_after: &str,
+        _fetch_owner: bool,
+        _start_after: &str,
     ) -> Result<ListObjectsV2Info> {
         let opts = ListPathOptions {
             bucket: bucket.to_string(),
@@ -247,9 +281,19 @@ impl StorageAPI for ECStore {
             ..Default::default()
         };
 
-        self.list_path(&opts).await?;
+        let info = self.list_path(&opts).await?;
 
-        Ok(ListObjectsV2Info::default())
+        warn!("list_objects_v2 info {:?}", info);
+
+        let v2 = ListObjectsV2Info {
+            is_truncated: info.is_truncated,
+            continuation_token: continuation_token.to_owned(),
+            next_continuation_token: info.next_marker,
+            objects: info.objects,
+            prefixes: info.prefixes,
+        };
+
+        Ok(v2)
     }
     async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
         let object = utils::path::encode_dir_object(object);

ecstore/src/store_api.rs

@@ -409,7 +409,7 @@ impl From<s3s::dto::CompletedPart> for CompletePart {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct ObjectInfo {
     pub bucket: String,
     pub name: String,
@@ -424,7 +424,7 @@ pub struct ObjectInfo {
     pub is_latest: bool,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct ListObjectsInfo {
     // Indicates whether the returned list objects response is truncated. A
     // value of true indicates that the list was truncated. The list can be truncated

rustfs/src/storage/ecfs.rs

@@ -10,7 +10,6 @@ use ecstore::store_api::PutObjReader;
 use ecstore::store_api::StorageAPI;
 use futures::pin_mut;
 use futures::{Stream, StreamExt};
-use futures_util::TryStreamExt;
 use http::HeaderMap;
 use s3s::dto::*;
 use s3s::s3_error;
@@ -20,9 +19,8 @@ use s3s::S3Result;
 use s3s::S3;
 use s3s::{S3Request, S3Response};
 use std::fmt::Debug;
-use std::pin::Pin;
 use std::str::FromStr;
-use std::task::Poll;
+use tracing::warn;
 use transform_stream::AsyncTryStream;
 
 use ecstore::error::Result;
@@ -258,13 +256,16 @@ impl S3 for FS {
             ..
         } = req.input;
 
-        let _object_infos = try_!(
+        let prefix = prefix.unwrap_or_default();
+        let delimiter = delimiter.unwrap_or_default();
+
+        let object_infos = try_!(
             self.store
                 .list_objects_v2(
                     &bucket,
-                    &prefix.unwrap_or_default(),
+                    &prefix,
                     &continuation_token.unwrap_or_default(),
-                    &delimiter.unwrap_or_default(),
+                    &delimiter,
                     max_keys.unwrap_or_default(),
                     fetch_owner.unwrap_or_default(),
                     &start_after.unwrap_or_default()
@@ -272,7 +273,41 @@ impl S3 for FS {
                 .await
         );
 
-        let output = ListObjectsV2Output { ..Default::default() };
+        warn!("object_infos {:?}", object_infos);
+
+        let objects: Vec<Object> = object_infos
+            .objects
+            .iter()
+            .filter(|v| !v.name.is_empty())
+            .map(|v| {
+                let mut obj = Object::default();
+                obj.key = Some(v.name.to_owned());
+                obj.last_modified = v.mod_time.map(Timestamp::from);
+                obj.size = Some(v.size as i64);
+
+                if fetch_owner.is_some_and(|v| v) {
+                    obj.owner = Some(Owner {
+                        display_name: Some("rustfs".to_owned()),
+                        id: Some("v0.1".to_owned()),
+                    });
+                }
+                obj
+            })
+            .collect();
+
+        let key_count = objects.len() as i32;
+
+        let output = ListObjectsV2Output {
+            key_count: Some(key_count),
+            max_keys: Some(key_count),
+            contents: Some(objects),
+            delimiter: Some(delimiter),
+            name: Some(bucket),
+            prefix: Some(prefix),
+            ..Default::default()
+        };
+
+        // let output = ListObjectsV2Output { ..Default::default() };
         Ok(S3Response::new(output))
     }