feat: Fs adds copy & link, TokioFs & TokioUringFs & `Monoio…

…` have been implemented
tonbo-io · Nov 11, 2024 · 6a34e6a · 6a34e6a
1 parent 8038993
commit 6a34e6a
Show file tree

Hide file tree

Showing 9 changed files with 375 additions and 16 deletions.
diff --git a/fusio-object-store/src/fs.rs b/fusio-object-store/src/fs.rs
@@ -1,10 +1,10 @@
-use std::sync::Arc;
+use std::{future::Future, sync::Arc};
 
 use async_stream::stream;
 use fusio::{
-    fs::{FileMeta, Fs, OpenOptions},
+    fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::Path,
-    Error,
+    Error, MaybeSend,
 };
 use futures_core::Stream;
 use futures_util::stream::StreamExt;
@@ -27,6 +27,10 @@ impl<O: ObjectStore> From<O> for S3Store<O> {
 impl<O: ObjectStore> Fs for S3Store<O> {
     type File = S3File<O>;
 
+    fn file_system(&self) -> FileSystemTag {
+        FileSystemTag::S3
+    }
+
     async fn open_options(&self, path: &Path, options: OpenOptions) -> Result<Self::File, Error> {
         if !options.truncate {
             return Err(Error::Unsupported {
@@ -64,4 +68,22 @@ impl<O: ObjectStore> Fs for S3Store<O> {
 
         Ok(())
     }
+
+    fn copy<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+        todo!()
+    }
+
+    fn link<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+        todo!()
+    }
 }
diff --git a/fusio-opendal/src/fs.rs b/fusio-opendal/src/fs.rs
@@ -1,7 +1,9 @@
+use std::future::Future;
+
 use fusio::{
-    fs::{FileMeta, Fs, OpenOptions},
+    fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::Path,
-    Error,
+    Error, MaybeSend,
 };
 use futures_core::Stream;
 use futures_util::TryStreamExt;
@@ -25,6 +27,10 @@ impl From<Operator> for OpendalFs {
 impl Fs for OpendalFs {
     type File = OpendalFile;
 
+    fn file_system(&self) -> FileSystemTag {
+        todo!()
+    }
+
     async fn open_options(&self, path: &Path, options: OpenOptions) -> Result<Self::File, Error> {
         OpendalFile::open(self.op.clone(), path.to_string(), options).await
     }
@@ -57,4 +63,22 @@ impl Fs for OpendalFs {
             .await
             .map_err(parse_opendal_error)
     }
+
+    fn copy<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+        todo!()
+    }
+
+    fn link<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+        todo!()
+    }
 }
diff --git a/fusio/src/fs/mod.rs b/fusio/src/fs/mod.rs
@@ -16,11 +16,21 @@ pub struct FileMeta {
     pub size: u64,
 }
 
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub enum FileSystemTag {
+    Local,
+    OPFS,
+    // TODO: Remote needs to check whether endpoint and other remote fs are consistent
+    S3,
+}
+
 pub trait Fs: MaybeSend + MaybeSync {
     //! This trait is used to abstract file system operations across different file systems.
 
     type File: Read + Write + MaybeSend + 'static;
 
+    fn file_system(&self) -> FileSystemTag;
+
     fn open(&self, path: &Path) -> impl Future<Output = Result<Self::File, Error>> {
         self.open_options(path, OpenOptions::default())
     }
@@ -39,4 +49,18 @@ pub trait Fs: MaybeSend + MaybeSync {
     ) -> impl Future<Output = Result<impl Stream<Item = Result<FileMeta, Error>>, Error>> + MaybeSend;
 
     fn remove(&self, path: &Path) -> impl Future<Output = Result<(), Error>> + MaybeSend;
+
+    fn copy<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend;
+
+    fn link<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend;
 }
diff --git a/fusio/src/impls/disk/monoio/fs.rs b/fusio/src/impls/disk/monoio/fs.rs
@@ -1,11 +1,11 @@
-use std::fs::create_dir_all;
+use std::{fs, fs::create_dir_all};
 
 use async_stream::stream;
 use futures_core::Stream;
 
 use super::MonoioFile;
 use crate::{
-    fs::{FileMeta, Fs, OpenOptions},
+    fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::{path_to_local, Path},
     Error,
 };
@@ -15,6 +15,10 @@ pub struct MonoIoFs;
 impl Fs for MonoIoFs {
     type File = MonoioFile;
 
+    fn file_system(&self) -> FileSystemTag {
+        FileSystemTag::Local
+    }
+
     async fn open_options(&self, path: &Path, options: OpenOptions) -> Result<Self::File, Error> {
         let local_path = path_to_local(path)?;
 
@@ -54,6 +58,33 @@ impl Fs for MonoIoFs {
     async fn remove(&self, path: &Path) -> Result<(), Error> {
         let path = path_to_local(path)?;
 
-        Ok(std::fs::remove_file(path)?)
+        Ok(fs::remove_file(path)?)
+    }
+
+    async fn copy<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        if self.file_system() == to_fs.file_system() {
+            let from = path_to_local(from)?;
+            let to = path_to_local(to)?;
+
+            fs::copy(&from, &to)?;
+        } else {
+            todo!()
+        }
+
+        Ok(())
+    }
+
+    async fn link<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        if self.file_system() != to_fs.file_system() {
+            return Err(Error::Unsupported {
+                message: "file system is inconsistent".to_string(),
+            });
+        }
+        let from = path_to_local(from)?;
+        let to = path_to_local(to)?;
+
+        fs::hard_link(&from, &to)?;
+
+        Ok(())
     }
 }
diff --git a/fusio/src/impls/disk/opfs/fs.rs b/fusio/src/impls/disk/opfs/fs.rs
@@ -1,3 +1,5 @@
+use std::future::Future;
+
 use async_stream::stream;
 use futures_core::Stream;
 use futures_util::StreamExt;
@@ -15,16 +17,20 @@ use super::OPFSFile;
 use crate::{
     disk::opfs::{promise, storage},
     error::wasm_err,
-    fs::{FileMeta, Fs, OpenOptions},
+    fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::Path,
-    Error,
+    Error, MaybeSend,
 };
 
 pub struct OPFS;
 
 impl Fs for OPFS {
     type File = OPFSFile;
 
+    fn file_system(&self) -> FileSystemTag {
+        FileSystemTag::Local
+    }
+
     async fn open_options(&self, path: &Path, options: OpenOptions) -> Result<Self::File, Error> {
         let segments: Vec<&str> = path.as_ref().trim_matches('/').split("/").collect();
 
@@ -97,6 +103,28 @@ impl Fs for OPFS {
             .map_err(wasm_err)?;
         Ok(())
     }
+
+    fn copy<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+        Err(Error::Unsupported {
+            message: "opfs does not support copy file".to_string(),
+        })
+    }
+
+    fn link<F: Fs>(
+        &self,
+        from: &Path,
+        to_fs: &F,
+        to: &Path,
+    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+        Err(Error::Unsupported {
+            message: "opfs does not support link file".to_string(),
+        })
+    }
 }
 
 impl OPFS {

diff --git a/fusio/src/impls/disk/tokio/fs.rs b/fusio/src/impls/disk/tokio/fs.rs
@@ -8,7 +8,7 @@ use tokio::{
 };
 
 use crate::{
-    fs::{FileMeta, Fs, OpenOptions},
+    fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::{path_to_local, Path},
     Error,
 };
@@ -18,6 +18,10 @@ pub struct TokioFs;
 impl Fs for TokioFs {
     type File = File;
 
+    fn file_system(&self) -> FileSystemTag {
+        FileSystemTag::Local
+    }
+
     async fn open_options(&self, path: &Path, options: OpenOptions) -> Result<Self::File, Error> {
         let local_path = path_to_local(path)?;
 
@@ -67,4 +71,31 @@ impl Fs for TokioFs {
         remove_file(&path).await?;
         Ok(())
     }
+
+    async fn copy<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        if self.file_system() == to_fs.file_system() {
+            let from = path_to_local(from)?;
+            let to = path_to_local(to)?;
+
+            tokio::fs::copy(&from, &to).await?;
+        } else {
+            todo!()
+        }
+
+        Ok(())
+    }
+
+    async fn link<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        if self.file_system() != to_fs.file_system() {
+            return Err(Error::Unsupported {
+                message: "file system is inconsistent".to_string(),
+            });
+        }
+        let from = path_to_local(from)?;
+        let to = path_to_local(to)?;
+
+        tokio::fs::hard_link(&from, &to).await?;
+
+        Ok(())
+    }
 }
diff --git a/fusio/src/impls/disk/tokio_uring/fs.rs b/fusio/src/impls/disk/tokio_uring/fs.rs
@@ -1,19 +1,25 @@
+use std::{fs, future::Future};
+
 use async_stream::stream;
 use futures_core::Stream;
 use tokio_uring::fs::{create_dir_all, remove_file};
 
 use crate::{
     disk::tokio_uring::TokioUringFile,
-    fs::{FileMeta, Fs, OpenOptions},
+    fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::{path_to_local, Path},
-    Error,
+    Error, MaybeSend,
 };
 
 pub struct TokioUringFs;
 
 impl Fs for TokioUringFs {
     type File = TokioUringFile;
 
+    fn file_system(&self) -> FileSystemTag {
+        FileSystemTag::Local
+    }
+
     async fn open_options(&self, path: &Path, options: OpenOptions) -> Result<Self::File, Error> {
         let local_path = path_to_local(path)?;
 
@@ -58,4 +64,31 @@ impl Fs for TokioUringFs {
 
         Ok(remove_file(path).await?)
     }
+
+    async fn copy<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        if self.file_system() == to_fs.file_system() {
+            let from = path_to_local(from)?;
+            let to = path_to_local(to)?;
+
+            fs::copy(&from, &to)?;
+        } else {
+            todo!()
+        }
+
+        Ok(())
+    }
+
+    async fn link<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        if self.file_system() != to_fs.file_system() {
+            return Err(Error::Unsupported {
+                message: "file system is inconsistent".to_string(),
+            });
+        }
+        let from = path_to_local(from)?;
+        let to = path_to_local(to)?;
+
+        fs::hard_link(&from, &to)?;
+
+        Ok(())
+    }
 }
diff --git a/fusio/src/impls/remotes/aws/fs.rs b/fusio/src/impls/remotes/aws/fs.rs
@@ -11,7 +11,7 @@ use url::Url;
 
 use super::{credential::AwsCredential, options::S3Options, S3Error, S3File};
 use crate::{
-    fs::{FileMeta, Fs, OpenOptions},
+    fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::Path,
     remotes::{
         aws::sign::Sign,
@@ -124,6 +124,10 @@ pub(super) struct AmazonS3Inner {
 impl Fs for AmazonS3 {
     type File = S3File;
 
+    fn file_system(&self) -> FileSystemTag {
+        FileSystemTag::S3
+    }
+
     async fn open_options(&self, path: &Path, _: OpenOptions) -> Result<Self::File, crate::Error> {
         Ok(S3File::new(self.clone(), path.clone()))
     }
@@ -233,6 +237,14 @@ impl Fs for AmazonS3 {
 
         Ok(())
     }
+
+    async fn copy<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        todo!()
+    }
+
+    async fn link<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
+        todo!()
+    }
 }
 
 #[derive(Debug, Deserialize)]