Fix tests

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

bindings/nodejs/src/lib.rs

@@ -668,8 +668,8 @@ impl Reader {
     /// TODO: change api into stream based.
     #[napi]
     pub async unsafe fn read(&mut self, mut buf: Buffer) -> Result<usize> {
-        let size = buf.len();
-        let bs = self.0.read(size).await.map_err(format_napi_error)?;
+        let buf = buf.as_mut();
+        let bs = self.0.read(buf.len()).await.map_err(format_napi_error)?;
         buf[..bs.len()].copy_from_slice(&bs);
         Ok(bs.len())
     }

core/src/raw/oio/read/futures_read.rs

@@ -58,7 +58,7 @@ where
     async fn read(&mut self, size: usize) -> Result<Bytes> {
         // Make sure buf has enough space.
         if self.buf.capacity() < size {
-            self.buf.reserve(size - self.buf.capacity());
+            self.buf.reserve(size);
         }
         let buf = self.buf.spare_capacity_mut();
         let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
@@ -70,14 +70,13 @@ where
 
         let n = self
             .inner
-            .read(read_buf.initialize_unfilled())
+            .read(read_buf.initialized_mut())
             .await
             .map_err(|err| {
                 new_std_io_error(err)
                     .with_operation(oio::ReadOperation::Read)
                     .with_context("source", "FuturesReader")
             })?;
-
         read_buf.set_filled(n);
 
         Ok(Bytes::copy_from_slice(read_buf.filled()))

core/src/raw/oio/read/tokio_read.rs

@@ -58,7 +58,7 @@ where
     async fn read(&mut self, size: usize) -> Result<Bytes> {
         // Make sure buf has enough space.
         if self.buf.capacity() < size {
-            self.buf.reserve(size - self.buf.capacity());
+            self.buf.reserve(size);
         }
         let buf = self.buf.spare_capacity_mut();
         let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
@@ -70,12 +70,12 @@ where
 
         let n = self
             .inner
-            .read(read_buf.initialize_unfilled())
+            .read(read_buf.initialized_mut())
             .await
             .map_err(|err| {
                 new_std_io_error(err)
                     .with_operation(oio::ReadOperation::Read)
-                    .with_context("source", "FuturesReader")
+                    .with_context("source", "TokioReader")
             })?;
         read_buf.set_filled(n);
 

core/src/services/ftp/util.rs

@@ -63,7 +63,7 @@ impl oio::Read for FtpReader {
 
         // Make sure buf has enough space.
         if self.buf.capacity() < size {
-            self.buf.reserve(size - self.buf.capacity());
+            self.buf.reserve(size);
         }
         let buf = self.buf.spare_capacity_mut();
         let mut read_buf: ReadBuf = ReadBuf::uninit(buf);

core/src/types/reader.rs

@@ -22,7 +22,7 @@ use std::task::ready;
 use std::task::Context;
 use std::task::Poll;
 
-use bytes::{Bytes, BytesMut};
+use bytes::{BufMut, Bytes, BytesMut};
 use futures::Stream;
 use tokio::io::ReadBuf;
 
@@ -151,7 +151,8 @@ impl Reader {
         }
 
         let mut bs = BytesMut::with_capacity(size);
-        bs.copy_from_slice(&bs1);
+        bs.put_slice(&bs1);
+
         let mut remaining = size - bs.len();
 
         loop {
@@ -163,7 +164,12 @@ impl Reader {
                         .with_context("actual", bs.len().to_string()),
                 );
             }
-            bs.copy_from_slice(&tmp);
+            bs.put_slice(&tmp);
+            debug_assert!(
+                tmp.len() <= remaining,
+                "read should not return more bytes than expected"
+            );
+
             remaining -= tmp.len();
             if remaining == 0 {
                 break;