feat(core): Streaming reading while chunk is not set

bindings/c/src/types.rs

@@ -58,10 +58,10 @@ impl opendal_bytes {
 }
 
 #[allow(clippy::from_over_into)]
-impl Into<bytes::Bytes> for opendal_bytes {
-    fn into(self) -> bytes::Bytes {
+impl Into<Buffer> for opendal_bytes {
+    fn into(self) -> Buffer {
         let slice = unsafe { std::slice::from_raw_parts(self.data, self.len) };
-        bytes::Bytes::copy_from_slice(slice)
+        Buffer::from(bytes::Bytes::copy_from_slice(slice))
     }
 }
 

core/Cargo.toml

@@ -56,6 +56,7 @@ tests = [
     "services-fs",
     "services-http",
     "services-memory",
+    "internal-tokio-rt",
     "services-s3",
 ]
 

core/src/layers/blocking.rs

@@ -282,8 +282,8 @@ impl<I> BlockingWrapper<I> {
 }
 
 impl<I: oio::Read + 'static> oio::BlockingRead for BlockingWrapper<I> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.handle.block_on(self.inner.read_at(offset, size))
+    fn read(&mut self) -> Result<Buffer> {
+        self.handle.block_on(self.inner.read())
     }
 }
 

core/src/layers/chaos.rs

@@ -171,19 +171,19 @@ impl<R> ChaosReader<R> {
 }
 
 impl<R: oio::Read> oio::Read for ChaosReader<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    async fn read(&mut self) -> Result<Buffer> {
         if self.i_feel_lucky() {
-            self.inner.read_at(offset, size).await
+            self.inner.read().await
         } else {
             Err(Self::unexpected_eof())
         }
     }
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for ChaosReader<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    fn read(&mut self) -> Result<Buffer> {
         if self.i_feel_lucky() {
-            self.inner.read_at(offset, size)
+            self.inner.read()
         } else {
             Err(Self::unexpected_eof())
         }

core/src/layers/complete.rs

@@ -587,39 +587,15 @@ pub type CompleteLister<A, P> =
 pub struct CompleteReader<R>(R);
 
 impl<R: oio::Read> oio::Read for CompleteReader<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        if size == 0 {
-            return Ok(Buffer::new());
-        }
-
-        let buf = self.0.read_at(offset, size).await?;
-        if buf.len() != size {
-            return Err(Error::new(
-                ErrorKind::RangeNotSatisfied,
-                "service didn't return the expected size",
-            )
-            .with_context("expect", size.to_string())
-            .with_context("actual", buf.len().to_string()));
-        }
+    async fn read(&mut self) -> Result<Buffer> {
+        let buf = self.0.read().await?;
         Ok(buf)
     }
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for CompleteReader<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        if size == 0 {
-            return Ok(Buffer::new());
-        }
-
-        let buf = self.0.read_at(offset, size)?;
-        if buf.len() != size {
-            return Err(Error::new(
-                ErrorKind::RangeNotSatisfied,
-                "service didn't return the expected size",
-            )
-            .with_context("expect", size.to_string())
-            .with_context("actual", buf.len().to_string()));
-        }
+    fn read(&mut self) -> Result<Buffer> {
+        let buf = self.0.read()?;
         Ok(buf)
     }
 }
@@ -742,7 +718,7 @@ mod tests {
         }
 
         async fn read(&self, _: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
-            Ok((RpRead::new(), Arc::new(bytes::Bytes::new())))
+            Ok((RpRead::new(), Box::new(bytes::Bytes::new())))
         }
 
         async fn write(&self, _: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {

core/src/layers/concurrent_limit.rs

@@ -250,14 +250,14 @@ impl<R> ConcurrentLimitWrapper<R> {
 }
 
 impl<R: oio::Read> oio::Read for ConcurrentLimitWrapper<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, size).await
+    async fn read(&mut self) -> Result<Buffer> {
+        self.inner.read().await
     }
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for ConcurrentLimitWrapper<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, size)
+    fn read(&mut self) -> Result<Buffer> {
+        self.inner.read()
     }
 }
 

core/src/layers/dtrace.rs

@@ -340,10 +340,10 @@ impl<R> DtraceLayerWrapper<R> {
 }
 
 impl<R: oio::Read> oio::Read for DtraceLayerWrapper<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    async fn read(&mut self) -> Result<Buffer> {
         let c_path = CString::new(self.path.clone()).unwrap();
         probe_lazy!(opendal, reader_read_start, c_path.as_ptr());
-        match self.inner.read_at(offset, size).await {
+        match self.inner.read().await {
             Ok(bs) => {
                 probe_lazy!(opendal, reader_read_ok, c_path.as_ptr(), bs.remaining());
                 Ok(bs)
@@ -357,11 +357,11 @@ impl<R: oio::Read> oio::Read for DtraceLayerWrapper<R> {
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for DtraceLayerWrapper<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    fn read(&mut self) -> Result<Buffer> {
         let c_path = CString::new(self.path.clone()).unwrap();
         probe_lazy!(opendal, blocking_reader_read_start, c_path.as_ptr());
         self.inner
-            .read_at(offset, size)
+            .read()
             .map(|bs| {
                 probe_lazy!(
                     opendal,

core/src/layers/error_context.rs

@@ -339,25 +339,21 @@ pub struct ErrorContextWrapper<T> {
 }
 
 impl<T: oio::Read> oio::Read for ErrorContextWrapper<T> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, size).await.map_err(|err| {
+    async fn read(&mut self) -> Result<Buffer> {
+        self.inner.read().await.map_err(|err| {
             err.with_operation(ReadOperation::Read)
                 .with_context("service", self.scheme)
                 .with_context("path", &self.path)
-                .with_context("offset", offset.to_string())
-                .with_context("size", size.to_string())
         })
     }
 }
 
 impl<T: oio::BlockingRead> oio::BlockingRead for ErrorContextWrapper<T> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, size).map_err(|err| {
+    fn read(&mut self) -> Result<Buffer> {
+        self.inner.read().map_err(|err| {
             err.with_operation(ReadOperation::BlockingRead)
                 .with_context("service", self.scheme)
                 .with_context("path", &self.path)
-                .with_context("offset", offset.to_string())
-                .with_context("size", size.to_string())
         })
     }
 }

core/src/layers/logging.rs

@@ -978,8 +978,8 @@ impl<R> Drop for LoggingReader<R> {
 }
 
 impl<R: oio::Read> oio::Read for LoggingReader<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        match self.inner.read_at(offset, size).await {
+    async fn read(&mut self) -> Result<Buffer> {
+        match self.inner.read().await {
             Ok(bs) => {
                 self.read
                     .fetch_add(bs.remaining() as u64, Ordering::Relaxed);
@@ -1014,8 +1014,8 @@ impl<R: oio::Read> oio::Read for LoggingReader<R> {
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for LoggingReader<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        match self.inner.read_at(offset, size) {
+    fn read(&mut self) -> Result<Buffer> {
+        match self.inner.read() {
             Ok(bs) => {
                 self.read
                     .fetch_add(bs.remaining() as u64, Ordering::Relaxed);

core/src/layers/madsim.rs

@@ -261,10 +261,9 @@ pub struct MadsimReader {
 }
 
 impl oio::Read for MadsimReader {
-    async fn read_at(&self, offset: u64, size: usize) -> crate::Result<Buffer> {
-        if let Some(ref data) = self.data {
-            let size = min(size, data.len());
-            Ok(data.clone().split_to(size).into())
+    async fn read(&mut self) -> crate::Result<Buffer> {
+        if let Some(data) = self.data.take() {
+            Ok(Buffer::from(data))
         } else {
             Ok(Buffer::new())
         }

core/src/layers/metrics.rs

@@ -747,10 +747,10 @@ impl<R> MetricWrapper<R> {
 }
 
 impl<R: oio::Read> oio::Read for MetricWrapper<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    async fn read(&mut self) -> Result<Buffer> {
         let start = Instant::now();
 
-        match self.inner.read_at(offset, size).await {
+        match self.inner.read().await {
             Ok(bs) => {
                 self.bytes_counter.increment(bs.remaining() as u64);
                 self.requests_duration_seconds
@@ -766,11 +766,11 @@ impl<R: oio::Read> oio::Read for MetricWrapper<R> {
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for MetricWrapper<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    fn read(&mut self) -> Result<Buffer> {
         let start = Instant::now();
 
         self.inner
-            .read_at(offset, size)
+            .read()
             .map(|bs| {
                 self.bytes_counter.increment(bs.remaining() as u64);
                 self.requests_duration_seconds

core/src/layers/minitrace.rs

@@ -292,16 +292,16 @@ impl<R> MinitraceWrapper<R> {
 
 impl<R: oio::Read> oio::Read for MinitraceWrapper<R> {
     #[trace(enter_on_poll = true)]
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, size).await
+    async fn read(&mut self) -> Result<Buffer> {
+        self.inner.read().await
     }
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for MinitraceWrapper<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    fn read(&mut self) -> Result<Buffer> {
         let _g = self.span.set_local_parent();
         let _span = LocalSpan::enter_with_local_parent(ReadOperation::BlockingRead.into_static());
-        self.inner.read_at(offset, size)
+        self.inner.read()
     }
 }
 

core/src/layers/oteltrace.rs

@@ -272,14 +272,14 @@ impl<R> OtelTraceWrapper<R> {
 }
 
 impl<R: oio::Read> oio::Read for OtelTraceWrapper<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, size).await
+    async fn read(&mut self) -> Result<Buffer> {
+        self.inner.read().await
     }
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for OtelTraceWrapper<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, size)
+    fn read(&mut self) -> Result<Buffer> {
+        self.inner.read()
     }
 }
 

core/src/layers/prometheus.rs

@@ -680,13 +680,13 @@ impl<R> PrometheusMetricWrapper<R> {
 }
 
 impl<R: oio::Read> oio::Read for PrometheusMetricWrapper<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    async fn read(&mut self) -> Result<Buffer> {
         let labels = self.stats.generate_metric_label(
             self.scheme.into_static(),
             Operation::Read.into_static(),
             &self.path,
         );
-        match self.inner.read_at(offset, size).await {
+        match self.inner.read().await {
             Ok(bytes) => {
                 self.stats
                     .bytes_total
@@ -703,14 +703,14 @@ impl<R: oio::Read> oio::Read for PrometheusMetricWrapper<R> {
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for PrometheusMetricWrapper<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    fn read(&mut self) -> Result<Buffer> {
         let labels = self.stats.generate_metric_label(
             self.scheme.into_static(),
             Operation::BlockingRead.into_static(),
             &self.path,
         );
         self.inner
-            .read_at(offset, size)
+            .read()
             .map(|bs| {
                 self.stats
                     .bytes_total

core/src/layers/prometheus_client.rs

@@ -533,10 +533,10 @@ impl<R> PrometheusMetricWrapper<R> {
 }
 
 impl<R: oio::Read> oio::Read for PrometheusMetricWrapper<R> {
-    async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    async fn read(&mut self) -> Result<Buffer> {
         let start = Instant::now();
 
-        match self.inner.read_at(offset, size).await {
+        match self.inner.read().await {
             Ok(bs) => {
                 self.metrics
                     .observe_bytes_total(self.scheme, self.op, bs.remaining());
@@ -554,10 +554,10 @@ impl<R: oio::Read> oio::Read for PrometheusMetricWrapper<R> {
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for PrometheusMetricWrapper<R> {
-    fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+    fn read(&mut self) -> Result<Buffer> {
         let start = Instant::now();
         self.inner
-            .read_at(offset, size)
+            .read()
             .map(|bs| {
                 self.metrics
                     .observe_bytes_total(self.scheme, self.op, bs.remaining());