[incubator-opendal] 01/01: feat(services/azblob): support sink

[su...@apache.org]

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

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

commit d8194551d29c0b8589a0f84e1ac4a798b1b23ad8
Author: suyanhanx <su...@gmail.com>
AuthorDate: Thu Jun 29 22:41:40 2023 +0800

    feat(services/azblob): support sink
    
    Signed-off-by: suyanhanx <su...@gmail.com>
---
 core/src/services/azblob/backend.rs |  1 +
 core/src/services/azblob/core.rs    |  2 +-
 core/src/services/azblob/writer.rs  | 29 +++++++++++++++++++++++------
 3 files changed, 25 insertions(+), 7 deletions(-)

diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index 7df1c52e9..b635f0203 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -531,6 +531,7 @@ impl Accessor for AzblobBackend {
                 read_with_override_content_disposition: true,
 
                 write: true,
+                write_can_sink: true,
                 write_with_cache_control: true,
                 write_with_content_type: true,
 
diff --git a/core/src/services/azblob/core.rs b/core/src/services/azblob/core.rs
index 9142cbae8..e15910828 100644
--- a/core/src/services/azblob/core.rs
+++ b/core/src/services/azblob/core.rs
@@ -244,7 +244,7 @@ impl AzblobCore {
     pub fn azblob_put_blob_request(
         &self,
         path: &str,
-        size: Option<usize>,
+        size: Option<u64>,
         content_type: Option<&str>,
         cache_control: Option<&str>,
         body: AsyncBody,
diff --git a/core/src/services/azblob/writer.rs b/core/src/services/azblob/writer.rs
index b751f5590..5cbf4e0d2 100644
--- a/core/src/services/azblob/writer.rs
+++ b/core/src/services/azblob/writer.rs
@@ -44,7 +44,7 @@ impl oio::Write for AzblobWriter {
     async fn write(&mut self, bs: Bytes) -> Result<()> {
         let mut req = self.core.azblob_put_blob_request(
             &self.path,
-            Some(bs.len()),
+            Some(bs.len() as u64),
             self.op.content_type(),
             self.op.cache_control(),
             AsyncBody::Bytes(bs),
@@ -65,11 +65,28 @@ impl oio::Write for AzblobWriter {
         }
     }
 
-    async fn sink(&mut self, _size: u64, _s: oio::Streamer) -> Result<()> {
-        Err(Error::new(
-            ErrorKind::Unsupported,
-            "Write::sink is not supported",
-        ))
+    async fn sink(&mut self, size: u64, s: oio::Streamer) -> Result<()> {
+        let mut req = self.core.azblob_put_blob_request(
+            &self.path,
+            Some(size),
+            self.op.content_type(),
+            self.op.cache_control(),
+            AsyncBody::Stream(s),
+        )?;
+
+        self.core.sign(&mut req).await?;
+
+        let resp = self.core.send(req).await?;
+
+        let status = resp.status();
+
+        match status {
+            StatusCode::CREATED | StatusCode::OK => {
+                resp.into_body().consume().await?;
+                Ok(())
+            }
+            _ => Err(parse_error(resp).await?),
+        }
     }
 
     async fn abort(&mut self) -> Result<()> {