[incubator-opendal] branch main updated: feat(services/sftp): add append support for sftp (#2297)

[xuanwo]

This is an automated email from the ASF dual-hosted git repository.

xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git


The following commit(s) were added to refs/heads/main by this push:
     new 31252d71 feat(services/sftp): add append support for sftp (#2297)
31252d71 is described below

commit 31252d71a2833aed79654721f8766726b8f4bbe0
Author: silver-ymz <yinmingz...@gmail.com>
AuthorDate: Tue May 23 18:43:52 2023 +0800

    feat(services/sftp): add append support for sftp (#2297)
    
    * feat(services/sftp): add append support for sftp
    
    Signed-off-by: silver-ymz <yinmingz...@gmail.com>
    
    * fix format
    
    Signed-off-by: silver-ymz <yinmingz...@gmail.com>
    
    ---------
    
    Signed-off-by: silver-ymz <yinmingz...@gmail.com>
---
 core/src/services/sftp/backend.rs | 23 ++++++++++++++++++++++-
 core/src/services/sftp/writer.rs  | 13 +++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/core/src/services/sftp/backend.rs 
b/core/src/services/sftp/backend.rs
index d15fba0e..9624a711 100644
--- a/core/src/services/sftp/backend.rs
+++ b/core/src/services/sftp/backend.rs
@@ -53,6 +53,7 @@ use crate::*;
 /// - [x] stat
 /// - [x] read
 /// - [x] write
+/// - [x] append
 /// - [x] create_dir
 /// - [x] delete
 /// - [x] copy
@@ -277,7 +278,7 @@ impl Accessor for SftpBackend {
     type BlockingReader = ();
     type Writer = SftpWriter;
     type BlockingWriter = ();
-    type Appender = ();
+    type Appender = SftpWriter;
     type Pager = Option<SftpPager>;
     type BlockingPager = ();
 
@@ -303,6 +304,7 @@ impl Accessor for SftpBackend {
 
                 copy: self.copyable,
                 rename: true,
+                append: true,
 
                 ..Default::default()
             });
@@ -387,6 +389,25 @@ impl Accessor for SftpBackend {
         Ok((RpWrite::new(), SftpWriter::new(file)))
     }
 
+    async fn append(&self, path: &str, _: OpAppend) -> Result<(RpAppend, 
Self::Appender)> {
+        if let Some((dir, _)) = path.rsplit_once('/') {
+            self.create_dir(dir, OpCreateDir::default()).await?;
+        }
+
+        let client = self.connect().await?;
+
+        let mut fs = client.fs();
+        fs.set_cwd(&self.root);
+        let path = fs.canonicalize(path).await?;
+
+        let mut option = client.options();
+        option.append(true).create(true);
+
+        let file = option.open(path).await?;
+
+        Ok((RpAppend::new(), SftpWriter::new(file)))
+    }
+
     async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
         let client = self.connect().await?;
 
diff --git a/core/src/services/sftp/writer.rs b/core/src/services/sftp/writer.rs
index 445a3457..5fa1c17d 100644
--- a/core/src/services/sftp/writer.rs
+++ b/core/src/services/sftp/writer.rs
@@ -53,3 +53,16 @@ impl oio::Write for SftpWriter {
         Ok(())
     }
 }
+
+#[async_trait]
+impl oio::Append for SftpWriter {
+    async fn append(&mut self, bs: Bytes) -> Result<()> {
+        self.file.write_all(&bs).await?;
+
+        Ok(())
+    }
+
+    async fn close(&mut self) -> Result<()> {
+        Ok(())
+    }
+}