refactor(core/raw): Align oio::BlockingRead API with oio::Read

bindings/c/src/reader.rs

@@ -15,8 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::io::Read;
-
 use ::opendal as core;
 
 use super::*;
@@ -51,12 +49,15 @@ impl opendal_reader {
         let buf = unsafe { std::slice::from_raw_parts_mut(buf, len) };
 
         let inner = unsafe { &mut *(*reader).inner };
-        let r = inner.read(buf);
+        let r = inner.read(buf.len());
         match r {
-            Ok(n) => opendal_result_reader_read {
-                size: n,
-                error: std::ptr::null_mut(),
-            },
+            Ok(bs) => {
+                buf[..bs.len()].copy_from_slice(&bs);
+                opendal_result_reader_read {
+                    size: bs.len(),
+                    error: std::ptr::null_mut(),
+                }
+            }
             Err(e) => opendal_result_reader_read {
                 size: 0,
                 error: opendal_error::new(

bindings/cpp/src/reader.rs

@@ -16,7 +16,6 @@
 // under the License.
 
 use anyhow::Result;
-use od::raw::oio::BlockingRead;
 use opendal as od;
 
 use super::ffi;
@@ -25,7 +24,9 @@ pub struct Reader(pub od::BlockingReader);
 
 impl Reader {
     pub fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
-        Ok(self.0.read(buf)?)
+        let bs = self.0.read(buf.len())?;
+        buf[..bs.len()].copy_from_slice(&bs);
+        Ok(bs.len())
     }
 
     pub fn seek(&mut self, offset: u64, dir: ffi::SeekFrom) -> Result<u64> {

bindings/nodejs/src/lib.rs

@@ -26,7 +26,6 @@ use std::time::Duration;
 
 use futures::TryStreamExt;
 use napi::bindgen_prelude::*;
-use opendal::raw::oio::BlockingRead;
 
 #[napi]
 pub struct Operator(opendal::Operator);
@@ -648,7 +647,10 @@ pub struct BlockingReader(opendal::BlockingReader);
 impl BlockingReader {
     #[napi]
     pub fn read(&mut self, mut buf: Buffer) -> Result<usize> {
-        self.0.read(buf.as_mut()).map_err(format_napi_error)
+        let buf = buf.as_mut();
+        let bs = self.0.read(buf.len()).map_err(format_napi_error)?;
+        buf[..bs.len()].copy_from_slice(&bs);
+        Ok(bs.len())
     }
 }
 

bindings/ocaml/src/operator/reader.rs

@@ -17,14 +17,14 @@
 
 use std::io;
 
-use opendal::raw::oio::BlockingRead;
-
 use super::*;
 
 #[ocaml::func]
 #[ocaml::sig("reader -> bytes -> (int, string) Result.t ")]
 pub fn reader_read(reader: &mut Reader, buf: &mut [u8]) -> Result<usize, String> {
-    map_res_error(reader.0.read(buf))
+    let bs = map_res_error(reader.0.read(buf.len()))?;
+    buf[..bs.len()].copy_from_slice(&bs);
+    Ok(bs.len())
 }
 
 #[ocaml::func]

bindings/python/src/file.rs

@@ -18,7 +18,6 @@
 // Remove this `allow` after <https://github.com/rust-lang/rust-clippy/issues/12039> fixed.
 #![allow(clippy::unnecessary_fallible_conversions)]
 
-use std::io::Read;
 use std::io::Seek;
 use std::io::SeekFrom;
 use std::io::Write;
@@ -77,11 +76,10 @@ impl File {
 
         let buffer = match size {
             Some(size) => {
-                let mut buffer = vec![0; size];
-                reader
-                    .read_exact(&mut buffer)
+                let bs = reader
+                    .read_exact(size)
                     .map_err(|err| PyIOError::new_err(err.to_string()))?;
-                buffer
+                bs.to_vec()
             }
             None => {
                 let mut buffer = Vec::new();

core/src/layers/blocking.rs

@@ -17,7 +17,7 @@
 
 use async_trait::async_trait;
 use bytes;
-use bytes::{BufMut, Bytes};
+use bytes::Bytes;
 use futures::future::poll_fn;
 use tokio::runtime::Handle;
 
@@ -288,25 +288,13 @@ impl<I> BlockingWrapper<I> {
 }
 
 impl<I: oio::Read + 'static> oio::BlockingRead for BlockingWrapper<I> {
-    fn read(&mut self, mut buf: &mut [u8]) -> Result<usize> {
-        let bs = self.handle.block_on(self.inner.read(buf.len()));
-        let bs = bs?;
-        buf.put_slice(&bs);
-        Ok(bs.len())
+    fn read(&mut self, limit: usize) -> Result<Bytes> {
+        self.handle.block_on(self.inner.read(limit))
     }
 
     fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64> {
         self.handle.block_on(self.inner.seek(pos))
     }
-
-    fn next(&mut self) -> Option<Result<Bytes>> {
-        let bs = self.handle.block_on(self.inner.read(4 * 1024 * 1024));
-        match bs {
-            Ok(bs) if bs.is_empty() => None,
-            Ok(bs) => Some(Ok(bs)),
-            Err(err) => Some(Err(err)),
-        }
-    }
 }
 
 impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {