feat(services/obs): support sink

core/src/services/obs/backend.rs

@@ -322,6 +322,7 @@ impl Accessor for ObsBackend {
                 read_with_if_none_match: true,
 
                 write: true,
+                write_can_sink: true,
                 write_with_content_type: true,
                 write_with_cache_control: true,
 

core/src/services/obs/core.rs

@@ -147,7 +147,7 @@ impl ObsCore {
     pub fn obs_put_object_request(
         &self,
         path: &str,
-        size: Option<usize>,
+        size: Option<u64>,
         content_type: Option<&str>,
         cache_control: Option<&str>,
         body: AsyncBody,

core/src/services/obs/writer.rs

@@ -37,17 +37,14 @@ impl ObsWriter {
     pub fn new(core: Arc<ObsCore>, op: OpWrite, path: String) -> Self {
         ObsWriter { core, op, path }
     }
-}
 
-#[async_trait]
-impl oio::Write for ObsWriter {
-    async fn write(&mut self, bs: Bytes) -> Result<()> {
+    async fn write_oneshot(&self, size: u64, body: AsyncBody) -> Result<()> {
         let mut req = self.core.obs_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 ObsWriter {
             _ => Err(parse_error(resp).await?),
         }
     }
+}
+
+#[async_trait]
+impl oio::Write for ObsWriter {
+    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<()> {