feat(services/cos): impl sink (apache#2587)

xxchan · Jul 1, 2023 · ca3b83b · ca3b83b
1 parent 4ada993
commit ca3b83b
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 12 deletions.
diff --git a/core/src/services/cos/backend.rs b/core/src/services/cos/backend.rs
@@ -266,6 +266,7 @@ impl Accessor for CosBackend {
                 read_with_if_none_match: true,
 
                 write: true,
+                write_can_sink: true,
                 write_with_content_type: true,
                 write_with_cache_control: true,
 

diff --git a/core/src/services/cos/core.rs b/core/src/services/cos/core.rs
@@ -147,7 +147,7 @@ impl CosCore {
     pub fn cos_put_object_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/cos/writer.rs b/core/src/services/cos/writer.rs
@@ -37,17 +37,14 @@ impl CosWriter {
     pub fn new(core: Arc<CosCore>, op: OpWrite, path: String) -> Self {
         CosWriter { core, op, path }
     }
-}
 
-#[async_trait]
-impl oio::Write for CosWriter {
-    async fn write(&mut self, bs: Bytes) -> Result<()> {
+    async fn write_oneshot(&self, size: u64, body: AsyncBody) -> Result<()> {
         let mut req = self.core.cos_put_object_request(
             &self.path,
-            Some(bs.len()),
+            Some(size),
             self.op.content_type(),
             self.op.cache_control(),
-            AsyncBody::Bytes(bs),
+            body,
         )?;
 
         self.core.sign(&mut req).await?;
@@ -64,12 +61,17 @@ impl oio::Write for CosWriter {
             _ => Err(parse_error(resp).await?),
         }
     }
+}
+
+#[async_trait]
+impl oio::Write for CosWriter {
+    async fn write(&mut self, bs: Bytes) -> Result<()> {
+        self.write_oneshot(bs.len() as u64, AsyncBody::Bytes(bs))
+            .await
+    }
 
-    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<()> {
+        self.write_oneshot(size, AsyncBody::Stream(s)).await
     }
 
     async fn abort(&mut self) -> Result<()> {