feat: impl copy for S3 & add fs::copy for different filesystems

fusio-object-store/src/fs.rs

@@ -69,21 +69,21 @@ 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!()
+    async fn copy(&self, from: &Path, to: &Path) -> Result<(), Error> {
+        let from = from.clone().into();
+        let to = to.clone().into();
+
+        self.inner
+            .copy(&from, &to)
+            .await
+            .map_err(BoxedError::from)?;
+
+        Ok(())
     }
 
-    fn link<F: Fs>(
-        &self,
-        from: &Path,
-        to_fs: &F,
-        to: &Path,
-    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
-        todo!()
+    async fn link<F: Fs>(&self, _: &Path, _: &F, _: &Path) -> Result<(), Error> {
+        Err(Error::Unsupported {
+            message: "s3 does not support link file".to_string(),
+        })
     }
 }

fusio-opendal/src/fs.rs

@@ -64,13 +64,8 @@ impl Fs for OpendalFs {
             .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 copy(&self, from: &Path, to: &Path) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+        self.op.copy(from.as_ref(), to.as_ref())
     }
 
     fn link<F: Fs>(

fusio/src/fs/mod.rs

@@ -3,12 +3,12 @@
 
 mod options;
 
-use std::future::Future;
+use std::{cmp, future::Future};
 
 use futures_core::Stream;
 pub use options::*;
 
-use crate::{path::Path, Error, MaybeSend, MaybeSync, Read, Write};
+use crate::{path::Path, Error, IoBufMut, MaybeSend, MaybeSync, Read, Write};
 
 #[derive(Debug)]
 pub struct FileMeta {
@@ -50,12 +50,7 @@ pub trait Fs: MaybeSend + MaybeSync {
 
     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 copy(&self, from: &Path, to: &Path) -> impl Future<Output = Result<(), Error>> + MaybeSend;
 
     fn link<F: Fs>(
         &self,
@@ -64,3 +59,37 @@ pub trait Fs: MaybeSend + MaybeSync {
         to: &Path,
     ) -> impl Future<Output = Result<(), Error>> + MaybeSend;
 }
+
+pub async fn copy<F, T>(from_fs: &F, from: &Path, to_fs: &T, to: &Path) -> Result<(), Error>
+where
+    F: Fs,
+    T: Fs,
+{
+    if from_fs.file_system() == to_fs.file_system() {
+        from_fs.copy(from, to).await?;
+        return Ok(());
+    }
+    let mut from_file = from_fs
+        .open_options(from, OpenOptions::default().read(true))
+        .await?;
+    let from_file_size = from_file.size().await? as usize;
+
+    let mut to_file = to_fs
+        .open_options(to, OpenOptions::default().create(true).write(true))
+        .await?;
+    let buf_size = cmp::min(from_file_size, 4 * 1024);
+    let mut buf = vec![0u8; buf_size];
+    let mut read_pos = 0u64;
+
+    while (read_pos as usize) < from_file_size - 1 {
+        let (result, _) = from_file.read_exact_at(buf.as_slice_mut(), read_pos).await;
+        result?;
+        read_pos += buf.len() as u64;
+
+        let (result, _) = to_file.write_all(buf.as_slice()).await;
+        result?;
+        buf.resize(buf_size, 0);
+    }
+
+    Ok(())
+}

fusio/src/impls/disk/monoio/fs.rs

@@ -61,15 +61,11 @@ impl Fs for MonoIoFs {
         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!()
-        }
+    async fn copy(&self, from: &Path, to: &Path) -> Result<(), Error> {
+        let from = path_to_local(from)?;
+        let to = path_to_local(to)?;
+
+        fs::copy(&from, &to)?;
 
         Ok(())
     }

fusio/src/impls/disk/opfs/fs.rs

@@ -104,23 +104,13 @@ impl Fs for OPFS {
         Ok(())
     }
 
-    fn copy<F: Fs>(
-        &self,
-        from: &Path,
-        to_fs: &F,
-        to: &Path,
-    ) -> impl Future<Output = Result<(), Error>> + MaybeSend {
+    async fn copy<F: Fs>(&self, from: &Path, to: &Path) -> Result<(), Error> {
         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 {
+    async fn link<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
         Err(Error::Unsupported {
             message: "opfs does not support link file".to_string(),
         })

fusio/src/impls/disk/tokio/fs.rs

@@ -72,15 +72,11 @@ impl Fs for TokioFs {
         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!()
-        }
+    async fn copy(&self, from: &Path, to: &Path) -> Result<(), Error> {
+        let from = path_to_local(from)?;
+        let to = path_to_local(to)?;
+
+        tokio::fs::copy(&from, &to).await?;
 
         Ok(())
     }

fusio/src/impls/disk/tokio_uring/fs.rs

@@ -65,15 +65,11 @@ 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!()
-        }
+    async fn copy(&self, from: &Path, to: &Path) -> Result<(), Error> {
+        let from = path_to_local(from)?;
+        let to = path_to_local(to)?;
+
+        fs::copy(&from, &to)?;
 
         Ok(())
     }

fusio/src/impls/remotes/aws/fs.rs

@@ -14,10 +14,13 @@ use crate::{
     fs::{FileMeta, FileSystemTag, Fs, OpenOptions},
     path::Path,
     remotes::{
-        aws::sign::Sign,
+        aws::{
+            multipart_upload::{MultipartUpload, UploadType},
+            sign::Sign,
+        },
         http::{DynHttpClient, HttpClient, HttpError},
     },
-    Error,
+    Error, Read,
 };
 
 pub struct AmazonS3Builder {
@@ -238,12 +241,29 @@ impl Fs for AmazonS3 {
         Ok(())
     }
 
-    async fn copy<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
-        todo!()
+    async fn copy(&self, from: &Path, to: &Path) -> Result<(), Error> {
+        let from_file = S3File::new(self.clone(), from.clone());
+        let from_file_size = from_file.size().await?;
+
+        let upload = MultipartUpload::new(self.clone(), to.clone());
+        upload
+            .upload_once(
+                from_file_size as usize,
+                UploadType::Copy {
+                    endpoint: self.inner.options.endpoint.clone(),
+                    from: from.clone(),
+                    body: Empty::<Bytes>::new(),
+                },
+            )
+            .await?;
+
+        Ok(())
     }
 
-    async fn link<F: Fs>(&self, from: &Path, to_fs: &F, to: &Path) -> Result<(), Error> {
-        todo!()
+    async fn link<F: Fs>(&self, _: &Path, _: &F, _: &Path) -> Result<(), Error> {
+        Err(Error::Unsupported {
+            message: "s3 does not support link file".to_string(),
+        })
     }
 }
 

fusio/src/impls/remotes/aws/multipart_upload.rs

@@ -27,6 +27,16 @@ pub(crate) struct MultipartUpload {
     path: Path,
 }
 
+pub enum UploadType<B> {
+    Write(B),
+    Copy {
+        endpoint: String,
+        from: Path,
+        // FIXME: for Empty
+        body: B,
+    },
+}
+
 impl MultipartUpload {
     pub fn new(fs: AmazonS3, path: Path) -> Self {
         Self { fs, path }
@@ -67,22 +77,42 @@ impl MultipartUpload {
         Self::check_response(response).await
     }
 
-    pub(crate) async fn upload_once<B>(&self, size: usize, body: B) -> Result<(), Error>
+    pub(crate) async fn upload_once<B>(
+        &self,
+        size: usize,
+        upload_type: UploadType<B>,
+    ) -> Result<(), Error>
     where
         B: Body<Data = Bytes> + Clone + Unpin + Send + Sync + 'static,
         B::Error: std::error::Error + Send + Sync + 'static,
     {
+        let (body, copy_from) = match upload_type {
+            UploadType::Write(body) => (body, None),
+            UploadType::Copy {
+                endpoint,
+                from: file_path,
+                body,
+            } => {
+                let from_url = format!(
+                    "{endpoint}/{}",
+                    utf8_percent_encode(file_path.as_ref(), &STRICT_PATH_ENCODE_SET)
+                );
+                (body, Some(from_url))
+            }
+        };
         let url = format!(
             "{}/{}",
             self.fs.as_ref().options.endpoint,
             utf8_percent_encode(self.path.as_ref(), &STRICT_PATH_ENCODE_SET)
         );
-        let request = Request::builder()
+        let mut builder = Request::builder()
             .uri(url)
             .method(Method::PUT)
-            .header(CONTENT_LENGTH, size)
-            .body(body)
-            .map_err(|e| Error::Other(e.into()))?;
+            .header(CONTENT_LENGTH, size);
+        if let Some(from_url) = copy_from {
+            builder = builder.header("x-amz-copy-source", from_url);
+        }
+        let request = builder.body(body).map_err(|e| Error::Other(e.into()))?;
         let _ = self.send_request(request).await?;
 
         Ok(())

fusio/src/impls/remotes/aws/writer.rs

@@ -2,11 +2,14 @@ use std::{mem, pin::Pin, sync::Arc};
 
 use bytes::{BufMut, BytesMut};
 use futures_util::{stream::FuturesOrdered, StreamExt};
-use http_body_util::Full;
+use http_body_util::{Empty, Full};
 
 use crate::{
     dynamic::MaybeSendFuture,
-    remotes::{aws::multipart_upload::MultipartUpload, serde::MultipartPart},
+    remotes::{
+        aws::multipart_upload::{MultipartUpload, UploadType},
+        serde::MultipartPart,
+    },
     Error, IoBuf, Write,
 };
 
@@ -90,7 +93,7 @@ impl Write for S3Writer {
                 let bytes = mem::replace(&mut self.buf, BytesMut::new()).freeze();
 
                 self.inner
-                    .upload_once(bytes.len(), Full::new(bytes))
+                    .upload_once(bytes.len(), UploadType::Write(Full::new(bytes)))
                     .await?;
             }
             return Ok(());

fusio/src/lib.rs

@@ -436,7 +436,6 @@ mod tests {
             src_fs
                 .copy(
                     &Path::from_absolute_path(&src_file_path)?,
-                    &dst_fs,
                     &Path::from_absolute_path(&dst_file_path)?,
                 )
                 .await?;