async list

Author: TD-Sky

core/Cargo.toml

@@ -192,7 +192,7 @@ services-oss = [
 services-pcloud = []
 services-persy = ["dep:persy", "internal-tokio-rt"]
 services-postgresql = ["dep:sqlx", "sqlx?/postgres"]
-services-redb = ["dep:redb", "internal-tokio-rt"]
+services-redb = ["dep:redb", "internal-tokio-rt", "dep:flume"]
 services-redis = ["dep:redis", "dep:bb8", "redis?/tokio-rustls-comp"]
 services-redis-native-tls = ["services-redis", "redis?/tokio-native-tls-comp"]
 services-rocksdb = ["dep:rocksdb", "internal-tokio-rt"]

core/src/services/redb/backend.rs

@@ -20,17 +20,19 @@ use std::sync::Arc;
 
 use tokio::task;
 
+use crate::raw::oio::HierarchyLister;
+use crate::raw::*;
+use crate::services::RedbConfig;
 use crate::Builder;
 use crate::Error;
 use crate::ErrorKind;
 use crate::Scheme;
-use crate::raw::oio::HierarchyLister;
-use crate::raw::*;
-use crate::services::RedbConfig;
 use crate::*;
 
 use super::core::RedbCore;
+use super::deleter::RedbDeleter;
 use super::error::*;
+use super::lister::RedbFilter;
 use super::lister::RedbLister;
 use super::writer::RedbWriter;
 
@@ -149,13 +151,15 @@ pub struct RedbBackend {
 impl Access for RedbBackend {
     type Reader = Buffer;
     type Writer = RedbWriter;
-    type Lister = ();
+    type Lister = HierarchyLister<RedbLister>;
+    type Deleter = oio::OneShotDeleter<RedbDeleter>;
     type BlockingReader = Buffer;
     type BlockingWriter = RedbWriter;
-    type BlockingLister = HierarchyLister<RedbLister>;
+    type BlockingLister = HierarchyLister<RedbFilter>;
+    type BlockingDeleter = oio::OneShotDeleter<RedbDeleter>;
 
     fn info(&self) -> Arc<AccessorInfo> {
-        let mut am = AccessorInfo::default();
+        let am = AccessorInfo::default();
         am.set_scheme(Scheme::Redb)
             .set_root(&self.core.root)
             .set_name(&self.core.datadir)
@@ -233,27 +237,30 @@ impl Access for RedbBackend {
         Ok((RpWrite::new(), RedbWriter::new(self.core.clone(), p)))
     }
 
-    async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
-        let cloned_self = self.clone();
-        let cloned_path = path.to_string();
+    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+        self.blocking_delete()
+    }
 
-        task::spawn_blocking(move || cloned_self.blocking_delete(cloned_path.as_str(), args))
-            .await
-            .map_err(new_task_join_error)
-            .and_then(|inner_result| inner_result)
+    fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+        Ok((
+            RpDelete::default(),
+            oio::OneShotDeleter::new(RedbDeleter::new(self.core.clone())),
+        ))
     }
 
-    fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
-        let p = build_abs_path(&self.core.root, path);
+    async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+        let pattern = build_abs_path(&self.core.root, path);
+        let range = self.core.iter()?;
+        let lister = RedbLister::new(RedbFilter::new(range, pattern));
+        let lister = HierarchyLister::new(lister, path, args.recursive());
 
-        self.core.delete(&p)?;
-        Ok(RpDelete::default())
+        Ok((RpList::default(), lister))
     }
 
     fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
         let pattern = build_abs_path(&self.core.root, path);
         let range = self.core.iter()?;
-        let lister = RedbLister::new(range, pattern);
+        let lister = RedbFilter::new(range, pattern);
         let lister = HierarchyLister::new(lister, path, args.recursive());
 
         Ok((RpList::default(), lister))

core/src/services/redb/deleter.rs

@@ -0,0 +1,42 @@
+use std::sync::Arc;
+
+use tokio::task;
+
+use crate::raw::build_abs_path;
+use crate::raw::new_task_join_error;
+use crate::raw::oio;
+use crate::raw::OpDelete;
+use crate::Result;
+
+use super::core::RedbCore;
+
+pub struct RedbDeleter {
+    core: Arc<RedbCore>,
+}
+
+impl RedbDeleter {
+    pub fn new(core: Arc<RedbCore>) -> Self {
+        RedbDeleter { core }
+    }
+}
+
+impl oio::OneShotDelete for RedbDeleter {
+    async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+        let p = build_abs_path(&self.core.root, &path);
+        let core = self.core.clone();
+
+        task::spawn_blocking(move || core.delete(&p))
+            .await
+            .map_err(new_task_join_error)
+            .and_then(|inner_result| inner_result)
+    }
+}
+
+impl oio::BlockingOneShotDelete for RedbDeleter {
+    fn blocking_delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+        let p = build_abs_path(&self.core.root, &path);
+
+        self.core.delete(&p)?;
+        Ok(())
+    }
+}

core/src/services/redb/lister.rs

@@ -1,22 +1,73 @@
+use tokio::task;
+
+use crate::raw::oio;
 use crate::EntryMode;
 use crate::Metadata;
 use crate::Result;
-use crate::raw::oio;
 
 use super::error::parse_storage_error;
 
+#[derive(Debug)]
 pub struct RedbLister {
+    receiver: flume::Receiver<Result<oio::Entry>>,
+}
+
+impl RedbLister {
+    pub fn new(mut filter: RedbFilter) -> Self {
+        let (tx, rx) = flume::bounded(1);
+
+        task::spawn_blocking(move || loop {
+            let Some(result) = filter.range.next() else {
+                break;
+            };
+
+            let (key, value) = match result {
+                Ok(pair) => pair,
+                Err(e) => {
+                    let e = parse_storage_error(e);
+                    if tx.send(Err(e)).is_err() {
+                        break;
+                    }
+                    continue;
+                }
+            };
+
+            let key = key.value();
+            let size = value.value().len() as u64;
+            if key.starts_with(&filter.pattern) {
+                let mode = EntryMode::from_path(key);
+                let entry = oio::Entry::new(key, Metadata::new(mode).with_content_length(size));
+                if tx.send(Ok(entry)).is_err() {
+                    break;
+                }
+            }
+        });
+
+        Self { receiver: rx }
+    }
+}
+
+impl oio::List for RedbLister {
+    async fn next(&mut self) -> Result<Option<oio::Entry>> {
+        match self.receiver.recv_async().await {
+            Ok(entry) => entry.map(Some),
+            Err(_) => Ok(None),
+        }
+    }
+}
+
+pub struct RedbFilter {
     range: redb::Range<'static, &'static str, &'static [u8]>,
     pattern: String,
 }
 
-impl RedbLister {
+impl RedbFilter {
     pub fn new(range: redb::Range<'static, &'static str, &'static [u8]>, pattern: String) -> Self {
         Self { range, pattern }
     }
 }
 
-impl oio::BlockingList for RedbLister {
+impl oio::BlockingList for RedbFilter {
     fn next(&mut self) -> Result<Option<oio::Entry>> {
         loop {
             let Some(result) = self.range.next() else {
@@ -27,11 +78,7 @@ impl oio::BlockingList for RedbLister {
             let key = key.value();
             let size = value.value().len() as u64;
             if key.starts_with(&self.pattern) {
-                let mode = if key.ends_with('/') {
-                    EntryMode::DIR
-                } else {
-                    EntryMode::FILE
-                };
+                let mode = EntryMode::from_path(key);
                 return Ok(Some(oio::Entry::new(
                     key,
                     Metadata::new(mode).with_content_length(size),

core/src/services/redb/mod.rs

@@ -18,6 +18,8 @@
 #[cfg(feature = "services-redb")]
 mod core;
 #[cfg(feature = "services-redb")]
+mod deleter;
+#[cfg(feature = "services-redb")]
 mod error;
 #[cfg(feature = "services-redb")]
 mod lister;

core/src/services/redb/writer.rs

@@ -2,10 +2,12 @@ use std::sync::Arc;
 
 use tokio::task;
 
-use crate::Buffer;
-use crate::Result;
 use crate::raw::new_task_join_error;
 use crate::raw::oio::{self, QueueBuf};
+use crate::Buffer;
+use crate::EntryMode;
+use crate::Metadata;
+use crate::Result;
 
 use super::core::RedbCore;
 
@@ -36,18 +38,23 @@ impl oio::Write for RedbWriter {
         Ok(())
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<Metadata> {
         let buf = self.buffer.clone().collect();
 
         let core = self.core.clone();
         let cloned_path = self.path.clone();
 
-        task::spawn_blocking(move || core.set(&cloned_path, &buf.to_vec()))
-            .await
-            .map_err(new_task_join_error)
-            .and_then(|inner_result| inner_result)?;
+        task::spawn_blocking(move || {
+            let value = buf.to_vec();
+            core.set(&cloned_path, &value)?;
 
-        Ok(())
+            let meta = Metadata::new(EntryMode::from_path(&cloned_path))
+                .with_content_length(value.len() as _);
+            Ok(meta)
+        })
+        .await
+        .map_err(new_task_join_error)
+        .and_then(|inner_result| inner_result)
     }
 
     async fn abort(&mut self) -> Result<()> {
@@ -62,9 +69,13 @@ impl oio::BlockingWrite for RedbWriter {
         Ok(())
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         let buf = self.buffer.clone().collect();
-        self.core.set(&self.path, &buf.to_vec())?;
-        Ok(())
+        let value = buf.to_vec();
+        self.core.set(&self.path, &value)?;
+
+        let meta =
+            Metadata::new(EntryMode::from_path(&self.path)).with_content_length(value.len() as _);
+        Ok(meta)
     }
 }