refactor(core): Implement RFC-2299 for presign_write_with (#2320)

refactor(core): Implement RFC-2299 for presign_write_with

Signed-off-by: clundro <[email protected]>

core/src/types/operator/operator.rs

@@ -1550,7 +1550,7 @@ impl Operator {
     /// curl -X PUT "https://s3.amazonaws.com/examplebucket/test.txt?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=access_key_id/20130721/us-east-1/s3/aws4_request&X-Amz-Date=20130721T201207Z&X-Amz-Expires=86400&X-Amz-SignedHeaders=host&X-Amz-Signature=<signature-value>" -d "Hello, World!"
     /// ```
     pub async fn presign_write(&self, path: &str, expire: Duration) -> Result<PresignedRequest> {
-        self.presign_write_with(path, OpWrite::new(), expire).await
+        self.presign_write_with(path, expire).await
     }
 
     /// Presign an operation for write with option described in OpenDAL [rfc-0661](../../docs/rfcs/0661-path-in-accessor.md)
@@ -1568,8 +1568,8 @@ impl Operator {
     ///
     /// #[tokio::main]
     /// async fn test(op: Operator) -> Result<()> {
-    ///     let args = OpWrite::new().with_content_type("text/csv");
-    ///     let signed_req = op.presign_write_with("test", args, Duration::from_secs(3600)).await?;
+    ///     let signed_req = op.presign_write_with("test", Duration::from_secs(3600))
+    ///                        .content_type("text/csv").await?;
     ///     let req = http::Request::builder()
     ///         .method(signed_req.method())
     ///         .uri(signed_req.uri())
@@ -1578,17 +1578,22 @@ impl Operator {
     /// #    Ok(())
     /// # }
     /// ```
-    pub async fn presign_write_with(
-        &self,
-        path: &str,
-        op: OpWrite,
-        expire: Duration,
-    ) -> Result<PresignedRequest> {
+    pub fn presign_write_with(&self, path: &str, expire: Duration) -> FuturePresignWrite {
         let path = normalize_path(path);
 
-        let op = OpPresign::new(op, expire);
-
-        let rp = self.inner().presign(&path, op).await?;
-        Ok(rp.into_presigned_request())
+        let fut = FuturePresignWrite(OperatorFuture::new(
+            self.inner().clone(),
+            path,
+            (OpWrite::default(), expire),
+            |inner, path, (args, dur)| {
+                let fut = async move {
+                    let op = OpPresign::new(args, dur);
+                    let rp = inner.presign(&path, op).await?;
+                    Ok(rp.into_presigned_request())
+                };
+                Box::pin(fut)
+            },
+        ));
+        fut
     }
 }

core/src/types/operator/operator_futures.rs

@@ -267,6 +267,57 @@ impl Future for FuturePresignRead {
     }
 }
 
+/// Future that generated by [`Operator::presign_read_with`].
+///
+/// Users can add more options by public functions provided by this struct.
+pub struct FuturePresignWrite(pub(crate) OperatorFuture<(OpWrite, Duration), PresignedRequest>);
+
+impl FuturePresignWrite {
+    /// Set the content length of op.
+    ///
+    /// If the content length is not set, the content length will be
+    /// calculated automatically by buffering part of data.
+    ///
+    pub fn content_length(mut self, v: u64) -> Self {
+        self.0 = self
+            .0
+            .map_args(|(args, dur)| (args.with_content_length(v), dur));
+        self
+    }
+
+    /// Set the content type of option
+    pub fn content_type(mut self, v: &str) -> Self {
+        self.0 = self
+            .0
+            .map_args(|(args, dur)| (args.with_content_type(v), dur));
+        self
+    }
+
+    /// Set the content disposition of option
+    pub fn content_disposition(mut self, v: &str) -> Self {
+        self.0 = self
+            .0
+            .map_args(|(args, dur)| (args.with_content_disposition(v), dur));
+        self
+    }
+
+    /// Set the content type of option
+    pub fn cache_control(mut self, v: &str) -> Self {
+        self.0 = self
+            .0
+            .map_args(|(args, dur)| (args.with_cache_control(v), dur));
+        self
+    }
+}
+
+impl Future for FuturePresignWrite {
+    type Output = Result<PresignedRequest>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.0.poll_unpin(cx)
+    }
+}
+
 /// Future that generated by [`Operator::read_with`].
 ///
 /// Users can add more options by public functions provided by this struct.