feat: Implement Writer::copy so user can copy from AsyncRead

core/src/raw/oio/into_stream/from_futures_reader.rs

@@ -0,0 +1,92 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::pin::Pin;
+use std::task::ready;
+use std::task::Context;
+use std::task::Poll;
+
+use bytes::BufMut;
+use bytes::Bytes;
+use bytes::BytesMut;
+use futures::AsyncRead;
+use tokio::io::ReadBuf;
+
+use crate::raw::*;
+use crate::*;
+
+// TODO: 64KiB is picked based on experiences, should be configurable
+const DEFAULT_CAPACITY: usize = 64 * 1024;
+
+/// Convert given futures reader into [`oio::Stream`].
+pub fn from_futures_reader<R>(r: R) -> FromFuturesReader<R>
+where
+    R: AsyncRead + Send + Sync + Unpin,
+{
+    FromFuturesReader {
+        inner: Some(r),
+        buf: BytesMut::new(),
+    }
+}
+
+pub struct FromFuturesReader<R> {
+    inner: Option<R>,
+    buf: BytesMut,
+}
+
+impl<S> oio::Stream for FromFuturesReader<S>
+where
+    S: AsyncRead + Send + Sync + Unpin,
+{
+    fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+        let reader = match self.inner.as_mut() {
+            Some(r) => r,
+            None => return Poll::Ready(None),
+        };
+
+        if self.buf.capacity() == 0 {
+            self.buf.reserve(DEFAULT_CAPACITY);
+        }
+
+        let dst = self.buf.spare_capacity_mut();
+        let mut buf = ReadBuf::uninit(dst);
+
+        // Safety: the buf must contains enough space for reading
+        unsafe { buf.assume_init(buf.capacity()) };
+
+        match ready!(Pin::new(reader).poll_read(cx, buf.initialized_mut())) {
+            Ok(0) => {
+                // Set inner to None while reaching EOF.
+                self.inner = None;
+                Poll::Ready(None)
+            }
+            Ok(n) => {
+                // Safety: read_exact makes sure this buffer has been filled.
+                unsafe { self.buf.advance_mut(n) }
+
+                let chunk = self.buf.split();
+                Poll::Ready(Some(Ok(chunk.freeze())))
+            }
+            Err(err) => Poll::Ready(Some(Err(Error::new(
+                ErrorKind::Unexpected,
+                "read data from reader into stream",
+            )
+            .set_temporary()
+            .set_source(err)))),
+        }
+    }
+}

core/src/raw/oio/into_stream/mod.rs

@@ -20,3 +20,6 @@
 
 mod from_futures_stream;
 pub use from_futures_stream::from_futures_stream;
+
+mod from_futures_reader;
+pub use from_futures_reader::from_futures_reader;

core/src/types/writer.rs

@@ -135,7 +135,55 @@ impl Writer {
             w.sink(size, s).await
         } else {
             unreachable!(
-                "writer state invalid while pipe_form, expect Idle, actual {}",
+                "writer state invalid while sink, expect Idle, actual {}",
+                self.state
+            );
+        }
+    }
+
+    /// Copy into writer.
+    ///
+    /// copy will read data from given reader and write them into writer
+    /// directly with only one constant in-memory buffer.
+    ///
+    /// # Notes
+    ///
+    /// - Copy doesn't support to be used with write concurrently.
+    /// - Copy doesn't support to be used without content length now.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// use std::io::Result;
+    ///
+    /// use bytes::Bytes;
+    /// use futures::stream;
+    /// use futures::StreamExt;
+    /// use opendal::Operator;
+    /// use futures::io::Cursor;
+    ///
+    /// #[tokio::main]
+    /// async fn copy_example(op: Operator) -> Result<()> {
+    ///     let mut w = op
+    ///         .writer_with("path/to/file")
+    ///         .content_length(4096)
+    ///         .await?;
+    ///     let reader = Cursor::new(vec![0;4096]);
+    ///     w.copy(4096, reader).await?;
+    ///     w.close().await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn copy<R>(&mut self, size: u64, read_from: R) -> Result<()>
+    where
+        R: futures::AsyncRead + Send + Sync + Unpin + 'static,
+    {
+        if let State::Idle(Some(w)) = &mut self.state {
+            let s = Box::new(oio::into_stream::from_futures_reader(read_from));
+            w.sink(size, s).await
+        } else {
+            unreachable!(
+                "writer state invalid while copy, expect Idle, actual {}",
                 self.state
             );
         }

core/tests/behavior/write.rs

@@ -37,7 +37,7 @@ use crate::*;
 pub fn behavior_write_tests(op: &Operator) -> Vec<Trial> {
     let cap = op.info().capability();
 
-    if !(cap.read && cap.write && cap.rename) {
+    if !(cap.read && cap.write) {
         return vec![];
     }
 
@@ -83,6 +83,7 @@ pub fn behavior_write_tests(op: &Operator) -> Vec<Trial> {
         test_remove_one_file,
         test_writer_write,
         test_writer_sink,
+        test_writer_copy,
         test_writer_abort,
         test_writer_futures_copy,
         test_fuzz_unsized_writer
@@ -1131,6 +1132,46 @@ pub async fn test_writer_sink(op: Operator) -> Result<()> {
     Ok(())
 }
 
+/// Reading data into writer
+pub async fn test_writer_copy(op: Operator) -> Result<()> {
+    let cap = op.info().capability();
+    if !(cap.write && cap.write_can_sink) {
+        return Ok(());
+    }
+
+    let path = uuid::Uuid::new_v4().to_string();
+    let size = 5 * 1024 * 1024; // write file with 5 MiB
+    let content_a = gen_fixed_bytes(size);
+    let content_b = gen_fixed_bytes(size);
+    let reader = Cursor::new(vec![content_a.clone(), content_b.clone()].concat());
+
+    let mut w = op
+        .writer_with(&path)
+        .content_length(2 * size as u64)
+        .await?;
+    w.copy(2 * size as u64, reader).await?;
+    w.close().await?;
+
+    let meta = op.stat(&path).await.expect("stat must succeed");
+    assert_eq!(meta.content_length(), (size * 2) as u64);
+
+    let bs = op.read(&path).await?;
+    assert_eq!(bs.len(), size * 2, "read size");
+    assert_eq!(
+        format!("{:x}", Sha256::digest(&bs[..size])),
+        format!("{:x}", Sha256::digest(content_a)),
+        "read content a"
+    );
+    assert_eq!(
+        format!("{:x}", Sha256::digest(&bs[size..])),
+        format!("{:x}", Sha256::digest(content_b)),
+        "read content b"
+    );
+
+    op.delete(&path).await.expect("delete must succeed");
+    Ok(())
+}
+
 /// Copy data from reader to writer
 pub async fn test_writer_futures_copy(op: Operator) -> Result<()> {
     let path = uuid::Uuid::new_v4().to_string();