This is an automated email from the ASF dual-hosted git repository. erickguan pushed a commit to branch fix-reader in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 380729d923cf12e5ba93c8dcfd5feacdd0580e8b Author: Erick Guan <[email protected]> AuthorDate: Sun Jul 5 11:40:14 2026 +0800 fix: reuse sftp read handles --- .../docs/rfcs/7660_move_read_range_to_reader.md | 101 +++++---- core/core/src/types/context/read.rs | 2 +- core/core/src/types/read/reader.rs | 236 +++++++++++++++++++++ core/services/sftp/src/backend.rs | 16 +- core/services/sftp/src/reader.rs | 99 +++------ 5 files changed, 333 insertions(+), 121 deletions(-) diff --git a/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md b/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md index 623906f60..1881e1780 100644 --- a/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md +++ b/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md @@ -6,19 +6,19 @@ # Summary Change OpenDAL's raw read contract so byte ranges are selected by the raw reader -instead of `Access::read`. +instead of `Service::read`. -Today `Access::read(path, OpRead { range, .. })` opens one concrete range +Today `Service::read(path, OpRead { range, .. })` opens one concrete range stream. This works well for HTTP object stores, where each range read is an independent request. It is a poor fit for services such as `fs`, `sftp`, `hdfs`, `hdfs-native`, and `monoiofs`, where the service may keep an opened handle and execute multiple cursor-independent range reads against it. The -current core reader planner still calls `Access::read` for every planned range, +current core reader planner still calls `Service::read` for every planned range, forcing those services to repeatedly open and close handles. -This RFC keeps the `Access::read` operation, but changes its contract: +This RFC keeps the `Service::read` operation, but changes its contract: -- `Access::read(path, OpRead)` creates a reusable raw `oio::Read` for the path +- `Service::read(path, OpRead)` creates a reusable raw `oio::Read` for the path and read condition set. - `oio::Read::open(range)` opens a range-scoped byte stream. - `oio::Read::read(range)` reads one bounded planner range into one `Buffer`. @@ -41,12 +41,12 @@ let parts = reader.fetch(vec![0..1024, 4096..8192]).await?; However, the current implementation plans these public calls by generating new raw range streams. `ReadGenerator::next_reader` and `ChunkedReader` both call -`Access::read(path, args.with_range(range))` for every range. For `s3`, `gcs`, +`Service::read(path, args.with_range(range))` for every range. For `s3`, `gcs`, and `azblob`, this is natural because each range is a new HTTP request. For handle based services, this means every range repeats open and seek work even when the target data can be represented by a reusable handle. -The root cause is that `Access::read` is modeled as a transport-level range +The root cause is that `Service::read` is modeled as a transport-level range request. OpenDAL needs a storage-neutral raw model: - Object stores can implement a raw reader that issues an independent ranged @@ -84,7 +84,7 @@ let ranges = reader.fetch(vec![0..1024, 4096..8192]).await?; The difference is internal. OpenDAL creates one reusable raw reader for `reader("path/to/file")`. After that, range operations are executed by this raw -reader instead of reopening through `Access::read` for every range. +reader instead of reopening through `Service::read` for every range. For HTTP object stores, the raw reader stores the path and read arguments, then sends a new ranged request for each range. Behavior and cost stay the same. @@ -98,17 +98,18 @@ a shared seek cursor. It must use cursor-independent reads such as `pread` / ## Raw read contract -`Access::read` remains the raw read operation: +`Service::read` remains the raw read operation: ```rust,ignore -pub trait Access: Send + Sync + Debug + Unpin + 'static { +pub trait Service: Send + Sync + Debug + Unpin + 'static { type Reader: oio::Read; fn read( &self, + ctx: &OperationContext, path: &str, args: OpRead, - ) -> impl Future<Output = Result<(RpRead, Self::Reader)>> + MaybeSend; + ) -> Result<Self::Reader>; } ``` @@ -117,7 +118,7 @@ The meaning changes: - `path` and `OpRead` select a path and read condition set. - The returned `Self::Reader` is a reusable raw read executor for that path and condition set. -- No range is selected by `Access::read`. +- No range is selected by `Service::read`. `OpRead` should no longer contain `BytesRange`: @@ -136,7 +137,7 @@ pub struct OpRead { `OpReader` remains an internal execution policy used by OpenDAL's public `Reader`. It controls `concurrent`, `chunk`, `gap`, `prefetch`, and -`content_length_hint`. It must not become part of the backend `Access::read` +`content_length_hint`. It must not become part of the backend `Service::read` contract. `ReadOptions` and `ReaderOptions` should be converted differently: @@ -277,10 +278,15 @@ does not require every service to optimize handle reuse immediately. range, and execution policy: ```rust,ignore -async fn read_inner(acc: Accessor, path: String, opts: ReadOptions) -> Result<Buffer> { +async fn read_inner( + ctx: OperationContext, + srv: Servicer, + path: String, + opts: ReadOptions, +) -> Result<Buffer> { let (range, op_read, op_reader) = opts.into(); - let (rp, raw_reader) = acc.read(&path, op_read).await?; - let reader = Reader::new(acc, path, raw_reader, op_reader, rp); + let raw_reader = srv.read(&ctx, &path, op_read)?; + let reader = Reader::new(ctx, srv, path, raw_reader, op_reader); reader.read(range.to_range()).await } ``` @@ -288,14 +294,19 @@ async fn read_inner(acc: Accessor, path: String, opts: ReadOptions) -> Result<Bu `Operator::reader_options` should create the reusable raw reader once: ```rust,ignore -async fn reader_inner(acc: Accessor, path: String, opts: ReaderOptions) -> Result<Reader> { +async fn reader_inner( + ctx: OperationContext, + srv: Servicer, + path: String, + opts: ReaderOptions, +) -> Result<Reader> { let (op_read, op_reader) = opts.into(); - let (rp, raw_reader) = acc.read(&path, op_read).await?; - Ok(Reader::new(acc, path, raw_reader, op_reader, rp)) + let raw_reader = srv.read(&ctx, &path, op_read)?; + Ok(Reader::new(ctx, srv, path, raw_reader, op_reader)) } ``` -For services where `Access::read` currently performs remote I/O, the returned +For services where `Service::read` currently performs remote I/O, the returned raw reader should be lazy. This preserves the lazy reader behavior: creating `op.reader(path)` should not require opening the underlying data request. @@ -305,7 +316,7 @@ raw reader should be lazy. This preserves the lazy reader behavior: creating ```rust,ignore pub struct ReadContext { - acc: Accessor, + srv: Servicer, path: String, args: OpRead, options: OpReader, @@ -314,20 +325,19 @@ pub struct ReadContext { } ``` -`acc`, `path`, and `args` are still useful for planning, especially when +`srv`, `path`, and `args` are still useful for planning, especially when OpenDAL needs `stat` to resolve unbounded ranges for chunked reads or `AsyncSeek` adapters. Actual raw range I/O should go through `reader.open` or -`reader.read`, not through repeated `acc.read` calls. +`reader.read`, not through repeated `srv.read` calls. `Reader::metadata()` remains cache-only. The cache is updated from `RpRead` -returned by `Access::read`, `open`, or `read`. Lazy backends may return -`RpRead::default()` from `Access::read` and fill metadata after the first raw -range operation. Public `Reader::fetch` observes metadata from the planned raw -`read` calls it executes. +returned by `open` or `read`. Lazy backends may return `RpRead::default()` from +the raw reader until the first raw range operation. Public `Reader::fetch` +observes metadata from the planned raw `read` calls it executes. ## Layers -Existing layers continue to wrap `Access::read`, but the wrapping point moves +Existing layers continue to wrap `Service::read`, but the wrapping point moves from a range stream to a reusable raw reader. ### Type eraser @@ -335,11 +345,10 @@ from a range stream to a reusable raw reader. The type eraser should erase raw readers using the chosen object-safe container: ```rust,ignore -async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, oio::Reader)> { +fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<oio::Reader> { self.inner - .read(path, args) - .await - .map(|(rp, r)| (rp, Box::new(r) as oio::Reader)) + .read(ctx, path, args) + .map(|r| Box::new(r) as oio::Reader) } ``` @@ -349,7 +358,7 @@ specific type alias. ### Complete layer -`CompleteLayer` currently reads `args.range().size()` in `Access::read` and +`CompleteLayer` previously read `args.range().size()` in `Service::read` and wraps the returned range stream. After this RFC, range size is only known when `open` or `read` is called. @@ -371,13 +380,13 @@ every range operation: ### Retry layer -`RetryLayer` currently stores `OpRead` and advances `args.range_mut()` after +`RetryLayer` previously stored `OpRead` and advanced `args.range_mut()` after each successful stream read. After this RFC, range progress must move from `OpRead` into range operation wrappers. The new retry model should be: -- retry `Access::read(path, OpRead)` while creating the reusable raw reader; +- retry `Service::read(path, OpRead)` while creating the reusable raw reader; - retry `raw_reader.open(range)` when opening a range stream fails before bytes are returned; - wrap the returned `ReadStream` and track bytes already emitted; @@ -391,7 +400,7 @@ for the same path and read condition set. ### Correctness check layer `CorrectnessCheckLayer` can keep checking `read_with_*` capabilities in -`Access::read`, because those options still belong to the path and read +`Service::read`, because those options still belong to the path and read condition set. ## Presign read @@ -420,7 +429,7 @@ OpReader)` and discards `OpReader`. Services like `s3`, `gcs`, `azblob`, `oss`, and `http` should return a raw reader that stores `core`, `path`, and `OpRead`. -Each `open(range)` sends the same request that the current `Access::read` sends +Each `open(range)` sends the same request that the current `Service::read` sends today. For example, S3's current logic: ```rust,ignore @@ -466,8 +475,8 @@ public `Reader` keep the same user-facing behavior. The raw API is breaking: -- third-party `Access` implementations must remove range handling from - `Access::read`; +- third-party `Service` implementations must remove range handling from + `Service::read`; - current range-scoped readers must become `ReadStream`; - services must return reusable raw readers; - layers must wrap reusable raw readers and range operation results. @@ -477,7 +486,7 @@ Migration can be staged mechanically: 1. Introduce `ReadStream` and the new raw `Read` trait shape. 2. Move `BytesRange` out of `OpRead`. 3. Update type erasure and core read context. -4. Convert object-store services by moving current `Access::read` request logic +4. Convert object-store services by moving current `Service::read` request logic into `open(range)`. 5. Convert memory-like services by slicing in `read(range)`. 6. Convert handle based services first with independent per-range opens, then @@ -500,9 +509,9 @@ The implementation should be considered complete when: # Drawbacks -- This is a breaking raw API change. Third-party `Access` implementations must +- This is a breaking raw API change. Third-party `Service` implementations must be updated. -- It touches many services because every `Access::read` implementation currently +- It touches many services because every `Service::read` implementation currently receives a range. - The rename from range-scoped `oio::Read` to `oio::ReadStream` is large but necessary to avoid overloading one trait with two meanings. @@ -525,7 +534,7 @@ progress, and request sizing into core and layers. OpenDAL already has `open(range)` for streaming and backpressure. The raw `read(range)` primitive should instead express a complete, core-planned, bounded range read. -## Add `Access::open_read_at` +## Add `Service::open_read_at` Rejected. It adds a parallel backend operation for an execution optimization. OpenDAL's raw operation surface should stay centered on semantic operations: @@ -552,10 +561,10 @@ ranges from it gives one stream two responsibilities and makes layer wrapping harder. The reusable raw reader should be the place that knows how to execute additional ranges. -## Add `ReadPlan` beside `Access::read` +## Add `ReadPlan` beside `Service::read` Rejected. It improves execution planning but keeps the old range-scoped -`Access::read` as the primary contract. This RFC fixes the raw contract instead +`Service::read` as the primary contract. This RFC fixes the raw contract instead of adding a second planning hook. # Prior art @@ -568,7 +577,7 @@ reduce extra `open` syscalls, and listed future `read_ranges` and native `read_at` support. The current raw contract did not fully realize that intent because range -selection stayed in `Access::read`. This RFC keeps RFC-4382's public direction +selection stayed in `Service::read`. This RFC keeps RFC-4382's public direction and moves the raw contract to the place where handle reuse and batch range reads can be implemented. diff --git a/core/core/src/types/context/read.rs b/core/core/src/types/context/read.rs index 0f8e64fb5..18cb028ca 100644 --- a/core/core/src/types/context/read.rs +++ b/core/core/src/types/context/read.rs @@ -35,7 +35,7 @@ pub struct ReadContext { args: OpRead, /// Options for the reader. options: OpReader, - /// Raw reader returned by [`Access::read`]. + /// Raw reader returned by [`Service::read`]. reader: oio::Reader, /// Complete object metadata observed from successful read opens. metadata: OnceLock<Metadata>, diff --git a/core/core/src/types/read/reader.rs b/core/core/src/types/read/reader.rs index 9fcfa691a..3c2d67734 100644 --- a/core/core/src/types/read/reader.rs +++ b/core/core/src/types/read/reader.rs @@ -541,6 +541,8 @@ impl Reader { mod tests { use std::sync::Arc; use std::sync::Mutex; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; use std::time::Duration; use bytes::Bytes; @@ -636,6 +638,240 @@ mod tests { )) } + #[derive(Debug, Clone)] + struct CountingService { + content: Bytes, + service_reads: Arc<AtomicUsize>, + raw_reads: Arc<Mutex<Vec<BytesRange>>>, + } + + impl CountingService { + fn new(content: Bytes) -> Self { + Self { + content, + service_reads: Arc::default(), + raw_reads: Arc::default(), + } + } + } + + impl Service for CountingService { + type Reader = CountingReader; + type Writer = (); + type Lister = (); + type Deleter = (); + type Copier = (); + + fn info(&self) -> ServiceInfo { + ServiceInfo::with_scheme("counting") + } + + fn capability(&self) -> Capability { + Capability { + stat: true, + read: true, + ..Default::default() + } + } + + async fn create_dir( + &self, + _: &OperationContext, + _: &str, + _: OpCreateDir, + ) -> Result<RpCreateDir> { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + async fn stat(&self, _: &OperationContext, _: &str, _: OpStat) -> Result<RpStat> { + Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(self.content.len() as u64), + )) + } + + fn read(&self, _: &OperationContext, _: &str, _: OpRead) -> Result<Self::Reader> { + self.service_reads.fetch_add(1, Ordering::Relaxed); + Ok(CountingReader { + content: self.content.clone(), + raw_reads: self.raw_reads.clone(), + }) + } + + fn write(&self, _: &OperationContext, _: &str, _: OpWrite) -> Result<Self::Writer> { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + fn delete(&self, _: &OperationContext) -> Result<Self::Deleter> { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + fn list(&self, _: &OperationContext, _: &str, _: OpList) -> Result<Self::Lister> { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + fn copy( + &self, + _: &OperationContext, + _: &str, + _: &str, + _: OpCopy, + _: OpCopier, + ) -> Result<Self::Copier> { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + async fn rename( + &self, + _: &OperationContext, + _: &str, + _: &str, + _: OpRename, + ) -> Result<RpRename> { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result<RpPresign> { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + } + + #[derive(Debug)] + struct CountingReader { + content: Bytes, + raw_reads: Arc<Mutex<Vec<BytesRange>>>, + } + + impl CountingReader { + fn slice(&self, range: BytesRange) -> Result<Bytes> { + let content_len = self.content.len(); + let (start, end) = match range { + BytesRange::Range { offset, size } => { + let start = offset as usize; + let end = size.map_or(content_len, |size| start + size as usize); + (start, end) + } + BytesRange::Suffix { size } => { + let start = content_len.saturating_sub(size as usize); + (start, content_len) + } + }; + + if end > content_len { + return Err(Error::new( + ErrorKind::RangeNotSatisfied, + "range exceeds content length", + )); + } + + Ok(self.content.slice(start..end)) + } + } + + impl oio::Read for CountingReader { + async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> { + self.raw_reads.lock().unwrap().push(range); + let content = self.slice(range)?; + Ok(( + RpRead::new( + Metadata::new(EntryMode::FILE).with_content_length(self.content.len() as u64), + ), + Box::new(CountingReadStream { content }) as Box<dyn oio::ReadStreamDyn>, + )) + } + + async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> { + self.raw_reads.lock().unwrap().push(range); + Ok(( + RpRead::new( + Metadata::new(EntryMode::FILE).with_content_length(self.content.len() as u64), + ), + Buffer::from(self.slice(range)?), + )) + } + } + + struct CountingReadStream { + content: Bytes, + } + + impl oio::ReadStream for CountingReadStream { + async fn read(&mut self) -> Result<Buffer> { + Ok(Buffer::from(std::mem::take(&mut self.content))) + } + } + + #[tokio::test] + async fn test_reader_reuses_raw_reader_for_planned_ranges() -> Result<()> { + let service = CountingService::new(Bytes::from_static(b"0123456789")); + let service_reads = service.service_reads.clone(); + let raw_reads = service.raw_reads.clone(); + let op = Operator::from_parts(OperationContext::default(), Arc::new(service)); + + let reader = op.reader_with("test_file").chunk(2).concurrent(2).await?; + assert_eq!(service_reads.load(Ordering::Relaxed), 1); + + let buf = reader.read(0..6).await?; + assert_eq!(buf.to_bytes(), b"012345".as_slice()); + + let bufs = reader.fetch(vec![6..8, 8..10]).await?; + assert_eq!(bufs[0].to_bytes(), b"67".as_slice()); + assert_eq!(bufs[1].to_bytes(), b"89".as_slice()); + + let mut async_reader = reader.clone().into_futures_async_read(0..10).await?; + let mut buf = [0; 3]; + futures::AsyncReadExt::read_exact(&mut async_reader, &mut buf) + .await + .unwrap(); + assert_eq!(&buf, b"012"); + futures::AsyncSeekExt::seek(&mut async_reader, std::io::SeekFrom::Start(4)) + .await + .unwrap(); + let mut buf = [0; 2]; + futures::AsyncReadExt::read_exact(&mut async_reader, &mut buf) + .await + .unwrap(); + assert_eq!(&buf, b"45"); + + assert_eq!(service_reads.load(Ordering::Relaxed), 1); + assert_eq!( + *raw_reads.lock().unwrap(), + vec![ + BytesRange::new(0, Some(2)), + BytesRange::new(2, Some(2)), + BytesRange::new(4, Some(2)), + BytesRange::new(6, Some(2)), + BytesRange::new(8, Some(2)), + BytesRange::new(0, Some(2)), + BytesRange::new(2, Some(2)), + BytesRange::new(4, Some(2)), + BytesRange::new(6, Some(2)), + ] + ); + + Ok(()) + } + #[tokio::test] async fn test_trait() -> Result<()> { let op = Operator::via_iter(services::MEMORY_SCHEME, [])?; diff --git a/core/services/sftp/src/backend.rs b/core/services/sftp/src/backend.rs index 5b8cb8ab9..256cbd1da 100644 --- a/core/services/sftp/src/backend.rs +++ b/core/services/sftp/src/backend.rs @@ -203,7 +203,7 @@ pub struct SftpBackend { } impl Service for SftpBackend { - type Reader = oio::StreamReader<SftpReader>; + type Reader = oio::PositionReader<SftpReader>; type Writer = SftpLazyWriter; type Lister = SftpLazyLister; type Deleter = oio::OneShotDeleter<SftpDeleter>; @@ -255,15 +255,11 @@ impl Service for SftpBackend { Ok(RpStat::new(meta)) } fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> { - let output: oio::StreamReader<SftpReader> = { - Ok(oio::StreamReader::new(SftpReader::new( - self.clone(), - path, - args, - ))) - }?; - - Ok(output) + Ok(oio::PositionReader::new(SftpReader::new( + self.clone(), + path, + args, + ))) } fn write(&self, ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> { diff --git a/core/services/sftp/src/reader.rs b/core/services/sftp/src/reader.rs index a2cf46881..39e97a053 100644 --- a/core/services/sftp/src/reader.rs +++ b/core/services/sftp/src/reader.rs @@ -22,7 +22,6 @@ use super::core::is_sftp_failure; use super::core::parse_sftp_error; use super::lister::SftpLister; use super::writer::SftpWriter; -use bytes::BytesMut; use fastpool::bounded; use opendal_core::raw::*; use opendal_core::*; @@ -30,59 +29,6 @@ use openssh_sftp_client::file::File; use std::io::SeekFrom; use tokio::io::AsyncSeekExt; -pub struct SftpReadStream { - /// Keep the connection alive while data stream is alive. - _conn: bounded::Object<Manager>, - - file: File, - chunk: usize, - size: Option<usize>, - read: usize, - buf: BytesMut, -} - -impl SftpReadStream { - pub fn new(conn: bounded::Object<Manager>, file: File, size: Option<u64>) -> Self { - Self { - _conn: conn, - file, - size: size.map(|v| v as usize), - chunk: 2 * 1024 * 1024, - read: 0, - buf: BytesMut::new(), - } - } -} - -impl oio::ReadStream for SftpReadStream { - async fn read(&mut self) -> Result<Buffer> { - if self.read >= self.size.unwrap_or(usize::MAX) { - return Ok(Buffer::new()); - } - - let size = if let Some(size) = self.size { - (size - self.read).min(self.chunk) - } else { - self.chunk - }; - self.buf.reserve(size); - - let Some(bytes) = self - .file - .read(size as u32, self.buf.split_off(0)) - .await - .map_err(parse_sftp_error)? - else { - return Ok(Buffer::new()); - }; - - self.read += bytes.len(); - self.buf = bytes; - let bs = self.buf.split(); - Ok(Buffer::from(bs.freeze())) - } -} - /// Reader returned by this backend. pub struct SftpReader { backend: SftpBackend, @@ -98,8 +44,22 @@ impl SftpReader { } } -impl oio::StreamRead for SftpReader { - async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> { +pub struct SftpReaderHandle { + /// Keep the connection alive while the file handle is cached by `PositionReader`. + _conn: bounded::Object<Manager>, + file: File, +} + +impl SftpReaderHandle { + pub(super) fn new(conn: bounded::Object<Manager>, file: File) -> Self { + Self { _conn: conn, file } + } +} + +impl oio::PositionRead for SftpReader { + type Handle = SftpReaderHandle; + + async fn open(&self) -> Result<Self::Handle> { let backend = &self.backend; let path = self.path.as_str(); @@ -110,21 +70,32 @@ impl oio::StreamRead for SftpReader { let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?; - let mut f = client + let f = client .open(path.as_path()) .await .map_err(parse_sftp_error)?; - if range.offset() != 0 { - f.seek(SeekFrom::Start(range.offset())) - .await - .map_err(new_std_io_error)?; + Ok(SftpReaderHandle::new(client, f)) + } + + async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> { + if size == 0 { + return Ok(Buffer::new()); } - let rp = RpRead::default(); - let stream = SftpReadStream::new(client, f, range.size()); + let mut file = handle.file.clone(); + file.seek(SeekFrom::Start(offset)) + .await + .map_err(new_std_io_error)?; - Ok((rp, Box::new(stream) as Box<dyn oio::ReadStreamDyn>)) + match file + .read(size as u32, bytes::BytesMut::with_capacity(size)) + .await + .map_err(parse_sftp_error)? + { + Some(bytes) => Ok(Buffer::from(bytes.freeze())), + None => Ok(Buffer::new()), + } } }
