You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/04/23 14:17:25 UTC
(opendal) 01/05: Refactor reader layout
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch refactor-dir
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 1332c393db4797588275cb35d348ace511b68397
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Apr 23 21:35:57 2024 +0800
Refactor reader layout
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/types/mod.rs | 4 +-
core/src/types/reader/buffer_stream.rs | 175 +++++++++++++
core/src/types/reader/futures_async_reader.rs | 159 ++++++++++++
core/src/types/reader/futures_bytes_stream.rs | 74 ++++++
core/src/types/reader/mod.rs | 27 +++
core/src/types/{ => reader}/reader.rs | 337 +-------------------------
6 files changed, 438 insertions(+), 338 deletions(-)
diff --git a/core/src/types/mod.rs b/core/src/types/mod.rs
index 8718cb590f..9e69ebe959 100644
--- a/core/src/types/mod.rs
+++ b/core/src/types/mod.rs
@@ -29,9 +29,7 @@ pub use metadata::Metadata;
pub use metadata::Metakey;
mod reader;
-pub use reader::into_futures_async_read::FuturesAsyncReader;
-pub use reader::into_futures_stream::FuturesBytesStream;
-pub use reader::Reader;
+pub use reader::*;
mod blocking_reader;
pub use blocking_reader::into_std_iterator::StdBytesIterator;
diff --git a/core/src/types/reader/buffer_stream.rs b/core/src/types/reader/buffer_stream.rs
new file mode 100644
index 0000000000..f5cd6cc0df
--- /dev/null
+++ b/core/src/types/reader/buffer_stream.rs
@@ -0,0 +1,175 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::pin::Pin;
+use std::sync::atomic::Ordering;
+use std::task::{Context, Poll};
+use std::{
+ ops::{Bound, RangeBounds},
+ sync::{atomic::AtomicBool, Arc},
+};
+
+use futures::stream::{self, Buffered, FusedStream, Iter};
+use futures::{Stream, StreamExt};
+
+use crate::raw::*;
+use crate::*;
+
+/// FutureIterator is an iterator that returns future generated by `Reader`'s
+/// read.
+struct FutureIterator {
+ r: oio::Reader,
+ chunk: Option<usize>,
+
+ offset: u64,
+ end: Option<u64>,
+ finished: Arc<AtomicBool>,
+}
+
+impl FutureIterator {
+ #[inline]
+ fn new(r: oio::Reader, chunk: Option<usize>, range: impl RangeBounds<u64>) -> Self {
+ let start = match range.start_bound().cloned() {
+ Bound::Included(start) => start,
+ Bound::Excluded(start) => start + 1,
+ Bound::Unbounded => 0,
+ };
+ let end = match range.end_bound().cloned() {
+ Bound::Included(end) => Some(end + 1),
+ Bound::Excluded(end) => Some(end),
+ Bound::Unbounded => None,
+ };
+
+ FutureIterator {
+ r,
+ chunk,
+ offset: start,
+ end,
+ finished: Arc::default(),
+ }
+ }
+}
+
+impl Iterator for FutureIterator {
+ type Item = BoxedFuture<'static, Result<Buffer>>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ if self.offset >= self.end.unwrap_or(u64::MAX) {
+ return None;
+ }
+ if self.finished.load(Ordering::Relaxed) {
+ return None;
+ }
+
+ let offset = self.offset;
+ // TODO: replace with services preferred chunk size.
+ let chunk = self.chunk.unwrap_or(4 * 1024 * 1024);
+ let limit = self
+ .end
+ .map(|end| ((end - self.offset) as usize).min(chunk))
+ .unwrap_or(chunk);
+ let finished = self.finished.clone();
+ let r = self.r.clone();
+
+ // Update self.offset before building future.
+ self.offset += limit as u64;
+ let fut = async move {
+ let buf = r.read_at_dyn(offset, limit).await?;
+ if buf.len() < limit || limit == 0 {
+ // Update finished marked if buf is less than limit.
+ finished.store(true, Ordering::Relaxed);
+ }
+ Ok(buf)
+ };
+
+ Some(Box::pin(fut))
+ }
+}
+
+/// BufferStream is a stream that returns [`Buffer`] from [`Reader`].
+///
+/// This stream will use concurrent read to fetch data from underlying storage.
+///
+/// # Notes
+///
+/// BufferStream uses `Buffered<Iter<ReadFutureIterator>>` internally,
+/// but we want to hide those details from users.
+pub struct BufferStream(Buffered<Iter<FutureIterator>>);
+
+impl BufferStream {
+ /// Create a new buffer stream from given reader.
+ #[inline]
+ pub fn new(r: oio::Reader, options: OpReader, range: impl RangeBounds<u64>) -> Self {
+ let iter = FutureIterator::new(r, options.chunk(), range);
+ let stream = stream::iter(iter).buffered(options.concurrent());
+
+ BufferStream(stream)
+ }
+}
+
+impl Stream for BufferStream {
+ type Item = Result<Buffer>;
+
+ #[inline]
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ self.0.poll_next_unpin(cx)
+ }
+}
+
+impl FusedStream for BufferStream {
+ fn is_terminated(&self) -> bool {
+ self.0.is_terminated()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use bytes::{Buf, Bytes};
+ use futures::TryStreamExt;
+ use pretty_assertions::assert_eq;
+
+ /// Make sure BufferStream implements `Unpin` and `Send`.
+ trait AssertTrait: Unpin + MaybeSend + 'static {}
+ impl AssertTrait for BufferStream {}
+
+ #[test]
+ fn test_future_iterator() {
+ let r: oio::Reader = Arc::new(Buffer::new());
+
+ let it = FutureIterator::new(r.clone(), Some(1), 1..3);
+ let futs: Vec<_> = it.collect();
+ assert_eq!(futs.len(), 2);
+ }
+
+ #[tokio::test]
+ async fn test_buffer_stream() {
+ let r: oio::Reader = Arc::new(Buffer::from(vec![
+ Bytes::from("Hello"),
+ Bytes::from("World"),
+ ]));
+
+ let s = BufferStream::new(r, OpReader::new(), 4..8);
+ let bufs: Vec<_> = s.try_collect().await.unwrap();
+ assert_eq!(bufs.len(), 1);
+ assert_eq!(bufs[0].chunk(), "o".as_bytes());
+
+ let buf: Buffer = bufs.into_iter().flatten().collect();
+ assert_eq!(buf.len(), 4);
+ assert_eq!(&buf.to_vec(), "oWor".as_bytes());
+ }
+}
diff --git a/core/src/types/reader/futures_async_reader.rs b/core/src/types/reader/futures_async_reader.rs
new file mode 100644
index 0000000000..73666a1a19
--- /dev/null
+++ b/core/src/types/reader/futures_async_reader.rs
@@ -0,0 +1,159 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::io;
+use std::io::SeekFrom;
+use std::ops::Range;
+use std::pin::Pin;
+use std::task::ready;
+use std::task::Context;
+use std::task::Poll;
+
+use bytes::Buf;
+use futures::AsyncBufRead;
+use futures::AsyncRead;
+use futures::AsyncSeek;
+use futures::StreamExt;
+
+use crate::raw::*;
+use crate::*;
+
+/// FuturesAsyncReader is the adapter of [`AsyncRead`], [`AsyncBufRead`] and [`AsyncSeek`]
+/// for [`Reader`].
+///
+/// Users can use this adapter in cases where they need to use [`AsyncRead`] related trait.
+///
+/// FuturesAsyncReader also implements [`Unpin`], [`Send`] and [`Sync`]
+pub struct FuturesAsyncReader {
+ r: oio::Reader,
+ options: OpReader,
+
+ stream: BufferStream,
+ buf: Buffer,
+ start: u64,
+ end: u64,
+ pos: u64,
+}
+
+impl FuturesAsyncReader {
+ /// NOTE: don't allow users to create FuturesAsyncReader directly.
+ #[inline]
+ pub(super) fn new(r: oio::Reader, options: OpReader, range: Range<u64>) -> Self {
+ let (start, end) = (range.start, range.end);
+ let stream = BufferStream::new(r.clone(), options.clone(), range);
+
+ FuturesAsyncReader {
+ r,
+ options,
+ stream,
+ buf: Buffer::new(),
+ start,
+ end,
+ pos: 0,
+ }
+ }
+}
+
+impl AsyncBufRead for FuturesAsyncReader {
+ fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+ let this = self.get_mut();
+ loop {
+ if this.buf.has_remaining() {
+ return Poll::Ready(Ok(this.buf.chunk()));
+ }
+
+ this.buf = match ready!(this.stream.poll_next_unpin(cx)) {
+ Some(Ok(buf)) => buf,
+ Some(Err(err)) => return Poll::Ready(Err(format_std_io_error(err))),
+ None => return Poll::Ready(Ok(&[])),
+ };
+ }
+ }
+
+ fn consume(mut self: Pin<&mut Self>, amt: usize) {
+ self.buf.advance(amt);
+ // Make sure buf has been dropped before starting new request.
+ // Otherwise, we will hold those bytes in memory until next
+ // buffer reaching.
+ if self.buf.is_empty() {
+ self.buf = Buffer::new();
+ }
+ self.pos += amt as u64;
+ }
+}
+
+/// TODO: implement vectored read.
+impl AsyncRead for FuturesAsyncReader {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<io::Result<usize>> {
+ let this = self.get_mut();
+
+ loop {
+ if this.buf.remaining() > 0 {
+ let size = this.buf.remaining().min(buf.len());
+ this.buf.copy_to_slice(&mut buf[..size]);
+ this.pos += size as u64;
+ return Poll::Ready(Ok(size));
+ }
+
+ this.buf = match ready!(this.stream.poll_next_unpin(cx)) {
+ Some(Ok(buf)) => buf,
+ Some(Err(err)) => return Poll::Ready(Err(format_std_io_error(err))),
+ None => return Poll::Ready(Ok(0)),
+ };
+ }
+ }
+}
+
+impl AsyncSeek for FuturesAsyncReader {
+ fn poll_seek(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ pos: SeekFrom,
+ ) -> Poll<io::Result<u64>> {
+ let new_pos = match pos {
+ SeekFrom::Start(pos) => pos as i64,
+ SeekFrom::End(pos) => self.end as i64 - self.start as i64 + pos,
+ SeekFrom::Current(pos) => self.pos as i64 + pos,
+ };
+
+ // Check if new_pos is negative.
+ if new_pos < 0 {
+ return Poll::Ready(Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ "invalid seek to a negative position",
+ )));
+ }
+
+ let new_pos = new_pos as u64;
+
+ if (self.pos..self.pos + self.buf.remaining() as u64).contains(&new_pos) {
+ let cnt = new_pos - self.pos;
+ self.buf.advance(cnt as _);
+ } else {
+ self.buf = Buffer::new();
+ self.stream =
+ BufferStream::new(self.r.clone(), self.options.clone(), new_pos..self.end);
+ }
+
+ self.pos = new_pos;
+ Poll::Ready(Ok(self.pos))
+ }
+}
diff --git a/core/src/types/reader/futures_bytes_stream.rs b/core/src/types/reader/futures_bytes_stream.rs
new file mode 100644
index 0000000000..019e03649a
--- /dev/null
+++ b/core/src/types/reader/futures_bytes_stream.rs
@@ -0,0 +1,74 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::io;
+use std::ops::RangeBounds;
+use std::pin::Pin;
+use std::task::ready;
+use std::task::Context;
+use std::task::Poll;
+
+use bytes::Bytes;
+use futures::Stream;
+use futures::StreamExt;
+
+use crate::raw::*;
+use crate::*;
+
+/// FuturesStream is the adapter of [`Stream`] for [`Reader`].
+///
+/// Users can use this adapter in cases where they need to use [`Stream`] trait.
+///
+/// FuturesStream also implements [`Unpin`], [`Send`] and [`Sync`].
+pub struct FuturesBytesStream {
+ stream: BufferStream,
+ buf: Buffer,
+}
+
+impl FuturesBytesStream {
+ /// NOTE: don't allow users to create FuturesStream directly.
+ #[inline]
+ pub(crate) fn new(r: oio::Reader, options: OpReader, range: impl RangeBounds<u64>) -> Self {
+ let stream = BufferStream::new(r, options, range);
+
+ FuturesBytesStream {
+ stream,
+ buf: Buffer::new(),
+ }
+ }
+}
+
+impl Stream for FuturesBytesStream {
+ type Item = io::Result<Bytes>;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let this = self.get_mut();
+
+ loop {
+ // Consume current buffer
+ if let Some(bs) = Iterator::next(&mut this.buf) {
+ return Poll::Ready(Some(Ok(bs)));
+ }
+
+ this.buf = match ready!(this.stream.poll_next_unpin(cx)) {
+ Some(Ok(buf)) => buf,
+ Some(Err(err)) => return Poll::Ready(Some(Err(format_std_io_error(err)))),
+ None => return Poll::Ready(None),
+ };
+ }
+ }
+}
diff --git a/core/src/types/reader/mod.rs b/core/src/types/reader/mod.rs
new file mode 100644
index 0000000000..e4b7c80b7f
--- /dev/null
+++ b/core/src/types/reader/mod.rs
@@ -0,0 +1,27 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#[allow(clippy::module_inception)]
+mod reader;
+pub use reader::Reader;
+
+mod buffer_stream;
+pub(crate) use buffer_stream::BufferStream;
+mod futures_async_reader;
+pub use futures_async_reader::FuturesAsyncReader;
+mod futures_bytes_stream;
+pub use futures_bytes_stream::FuturesBytesStream;
diff --git a/core/src/types/reader.rs b/core/src/types/reader/reader.rs
similarity index 50%
rename from core/src/types/reader.rs
rename to core/src/types/reader/reader.rs
index 77d70a47d3..8fab33e9a8 100644
--- a/core/src/types/reader.rs
+++ b/core/src/types/reader/reader.rs
@@ -163,8 +163,8 @@ impl Reader {
/// This API can be public but we are not sure if it's useful for users.
/// And the name `BufferStream` is not good enough to expose to users.
/// Let's keep it inside for now.
- fn into_stream(self, range: impl RangeBounds<u64>) -> into_stream::BufferStream {
- into_stream::BufferStream::new(self.inner, self.options, range)
+ fn into_stream(self, range: impl RangeBounds<u64>) -> BufferStream {
+ BufferStream::new(self.inner, self.options, range)
}
/// Convert reader into [`FuturesAsyncReader`] which implements [`futures::AsyncRead`],
@@ -185,339 +185,6 @@ impl Reader {
}
}
-pub mod into_stream {
- use std::pin::Pin;
- use std::sync::atomic::Ordering;
- use std::task::{Context, Poll};
- use std::{
- ops::{Bound, RangeBounds},
- sync::{atomic::AtomicBool, Arc},
- };
-
- use futures::stream::{self, Buffered, Iter};
- use futures::{Stream, StreamExt};
-
- use crate::raw::*;
- use crate::*;
-
- /// ReadFutureIterator is an iterator that returns future of [`Buffer`] from [`Reader`].
- struct ReadFutureIterator {
- r: oio::Reader,
- chunk: Option<usize>,
-
- offset: u64,
- end: Option<u64>,
- finished: Arc<AtomicBool>,
- }
-
- impl ReadFutureIterator {
- #[inline]
- fn new(r: oio::Reader, chunk: Option<usize>, range: impl RangeBounds<u64>) -> Self {
- let start = match range.start_bound().cloned() {
- Bound::Included(start) => start,
- Bound::Excluded(start) => start + 1,
- Bound::Unbounded => 0,
- };
- let end = match range.end_bound().cloned() {
- Bound::Included(end) => Some(end + 1),
- Bound::Excluded(end) => Some(end),
- Bound::Unbounded => None,
- };
-
- ReadFutureIterator {
- r,
- chunk,
- offset: start,
- end,
- finished: Arc::default(),
- }
- }
- }
-
- impl Iterator for ReadFutureIterator {
- type Item = BoxedFuture<'static, Result<Buffer>>;
-
- fn next(&mut self) -> Option<Self::Item> {
- if self.offset >= self.end.unwrap_or(u64::MAX) {
- return None;
- }
- if self.finished.load(Ordering::Relaxed) {
- return None;
- }
-
- let offset = self.offset;
- // TODO: replace with services preferred chunk size.
- let chunk = self.chunk.unwrap_or(4 * 1024 * 1024);
- let limit = self
- .end
- .map(|end| ((end - self.offset) as usize).min(chunk))
- .unwrap_or(chunk);
- let finished = self.finished.clone();
- let r = self.r.clone();
-
- // Update self.offset before building future.
- self.offset += limit as u64;
- let fut = async move {
- let buf = r.read_at_dyn(offset, limit).await?;
- if buf.len() < limit || limit == 0 {
- // Update finished marked if buf is less than limit.
- finished.store(true, Ordering::Relaxed);
- }
- Ok(buf)
- };
-
- Some(Box::pin(fut))
- }
- }
-
- /// BufferStream is a stream that returns [`Buffer`] from [`Reader`].
- ///
- /// This stream will use concurrent read to fetch data from underlying storage.
- ///
- /// # Notes
- ///
- /// BufferStream uses `Buffered<Iter<ReadFutureIterator>>` internally,
- /// but we want to hide those details from users.
- pub struct BufferStream(Buffered<Iter<ReadFutureIterator>>);
-
- impl BufferStream {
- /// Create a new buffer stream from given reader.
- #[inline]
- pub fn new(r: oio::Reader, options: OpReader, range: impl RangeBounds<u64>) -> Self {
- let iter = ReadFutureIterator::new(r, options.chunk(), range);
- let stream = stream::iter(iter).buffered(options.concurrent());
-
- BufferStream(stream)
- }
- }
-
- impl Stream for BufferStream {
- type Item = Result<Buffer>;
-
- #[inline]
- fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
- self.0.poll_next_unpin(cx)
- }
- }
-
- #[cfg(test)]
- mod tests {
- use super::*;
- use crate::raw::MaybeSend;
-
- trait AssertTrait: Unpin + MaybeSend + 'static {}
- impl AssertTrait for BufferStream {}
- }
-}
-
-pub mod into_futures_async_read {
- use std::io;
- use std::io::SeekFrom;
- use std::ops::Range;
- use std::pin::Pin;
- use std::task::ready;
- use std::task::Context;
- use std::task::Poll;
-
- use bytes::Buf;
- use futures::AsyncBufRead;
- use futures::AsyncRead;
- use futures::AsyncSeek;
- use futures::StreamExt;
-
- use crate::raw::*;
- use crate::*;
-
- use super::into_stream::BufferStream;
-
- /// FuturesAsyncReader is the adapter of [`AsyncRead`], [`AsyncBufRead`] and [`AsyncSeek`]
- /// for [`Reader`].
- ///
- /// Users can use this adapter in cases where they need to use [`AsyncRead`] related trait.
- ///
- /// FuturesAsyncReader also implements [`Unpin`], [`Send`] and [`Sync`]
- pub struct FuturesAsyncReader {
- r: oio::Reader,
- options: OpReader,
-
- stream: BufferStream,
- buf: Buffer,
- start: u64,
- end: u64,
- pos: u64,
- }
-
- impl FuturesAsyncReader {
- /// NOTE: don't allow users to create FuturesAsyncReader directly.
- #[inline]
- pub(super) fn new(r: oio::Reader, options: OpReader, range: Range<u64>) -> Self {
- let (start, end) = (range.start, range.end);
- let stream = BufferStream::new(r.clone(), options.clone(), range);
-
- FuturesAsyncReader {
- r,
- options,
- stream,
- buf: Buffer::new(),
- start,
- end,
- pos: 0,
- }
- }
- }
-
- impl AsyncBufRead for FuturesAsyncReader {
- fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
- let this = self.get_mut();
- loop {
- if this.buf.has_remaining() {
- return Poll::Ready(Ok(this.buf.chunk()));
- }
-
- this.buf = match ready!(this.stream.poll_next_unpin(cx)) {
- Some(Ok(buf)) => buf,
- Some(Err(err)) => return Poll::Ready(Err(format_std_io_error(err))),
- None => return Poll::Ready(Ok(&[])),
- };
- }
- }
-
- fn consume(mut self: Pin<&mut Self>, amt: usize) {
- self.buf.advance(amt);
- // Make sure buf has been dropped before starting new request.
- // Otherwise, we will hold those bytes in memory until next
- // buffer reaching.
- if self.buf.is_empty() {
- self.buf = Buffer::new();
- }
- self.pos += amt as u64;
- }
- }
-
- /// TODO: implement vectored read.
- impl AsyncRead for FuturesAsyncReader {
- fn poll_read(
- self: Pin<&mut Self>,
- cx: &mut Context<'_>,
- buf: &mut [u8],
- ) -> Poll<io::Result<usize>> {
- let this = self.get_mut();
-
- loop {
- if this.buf.remaining() > 0 {
- let size = this.buf.remaining().min(buf.len());
- this.buf.copy_to_slice(&mut buf[..size]);
- this.pos += size as u64;
- return Poll::Ready(Ok(size));
- }
-
- this.buf = match ready!(this.stream.poll_next_unpin(cx)) {
- Some(Ok(buf)) => buf,
- Some(Err(err)) => return Poll::Ready(Err(format_std_io_error(err))),
- None => return Poll::Ready(Ok(0)),
- };
- }
- }
- }
-
- impl AsyncSeek for FuturesAsyncReader {
- fn poll_seek(
- mut self: Pin<&mut Self>,
- _: &mut Context<'_>,
- pos: SeekFrom,
- ) -> Poll<io::Result<u64>> {
- let new_pos = match pos {
- SeekFrom::Start(pos) => pos as i64,
- SeekFrom::End(pos) => self.end as i64 - self.start as i64 + pos,
- SeekFrom::Current(pos) => self.pos as i64 + pos,
- };
-
- // Check if new_pos is negative.
- if new_pos < 0 {
- return Poll::Ready(Err(io::Error::new(
- io::ErrorKind::InvalidInput,
- "invalid seek to a negative position",
- )));
- }
-
- let new_pos = new_pos as u64;
-
- if (self.pos..self.pos + self.buf.remaining() as u64).contains(&new_pos) {
- let cnt = new_pos - self.pos;
- self.buf.advance(cnt as _);
- } else {
- self.buf = Buffer::new();
- self.stream =
- BufferStream::new(self.r.clone(), self.options.clone(), new_pos..self.end);
- }
-
- self.pos = new_pos;
- Poll::Ready(Ok(self.pos))
- }
- }
-}
-
-pub mod into_futures_stream {
- use std::io;
- use std::ops::RangeBounds;
- use std::pin::Pin;
- use std::task::ready;
- use std::task::Context;
- use std::task::Poll;
-
- use bytes::Bytes;
- use futures::Stream;
- use futures::StreamExt;
-
- use super::into_stream::BufferStream;
- use crate::raw::*;
- use crate::*;
-
- /// FuturesStream is the adapter of [`Stream`] for [`Reader`].
- ///
- /// Users can use this adapter in cases where they need to use [`Stream`] trait.
- ///
- /// FuturesStream also implements [`Unpin`], [`Send`] and [`Sync`].
- pub struct FuturesBytesStream {
- stream: BufferStream,
- buf: Buffer,
- }
-
- impl FuturesBytesStream {
- /// NOTE: don't allow users to create FuturesStream directly.
- #[inline]
- pub(crate) fn new(r: oio::Reader, options: OpReader, range: impl RangeBounds<u64>) -> Self {
- let stream = BufferStream::new(r, options, range);
-
- FuturesBytesStream {
- stream,
- buf: Buffer::new(),
- }
- }
- }
-
- impl Stream for FuturesBytesStream {
- type Item = io::Result<Bytes>;
-
- fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
- let this = self.get_mut();
-
- loop {
- // Consume current buffer
- if let Some(bs) = Iterator::next(&mut this.buf) {
- return Poll::Ready(Some(Ok(bs)));
- }
-
- this.buf = match ready!(this.stream.poll_next_unpin(cx)) {
- Some(Ok(buf)) => buf,
- Some(Err(err)) => return Poll::Ready(Some(Err(format_std_io_error(err)))),
- None => return Poll::Ready(None),
- };
- }
- }
- }
-}
-
#[cfg(test)]
mod tests {
use rand::rngs::ThreadRng;