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());