diff options
author | Alex Auvolat <alex@adnab.me> | 2024-02-23 11:46:57 +0100 |
---|---|---|
committer | Alex Auvolat <alex@adnab.me> | 2024-02-23 11:46:57 +0100 |
commit | 9b41f4ff2016b099ee706747828f5828f8c96b1c (patch) | |
tree | 3c9e06d8d0ca7162e97e765bbe8091a5eaf676fd /src/net | |
parent | 93552b9275a6a83653353e27c0ee0c64c7a23d59 (diff) | |
download | garage-9b41f4ff2016b099ee706747828f5828f8c96b1c.tar.gz garage-9b41f4ff2016b099ee706747828f5828f8c96b1c.zip |
[refactor-block] move read_stream_to_end to garage_net
Diffstat (limited to 'src/net')
-rw-r--r-- | src/net/bytes_buf.rs | 13 | ||||
-rw-r--r-- | src/net/stream.rs | 11 |
2 files changed, 24 insertions, 0 deletions
diff --git a/src/net/bytes_buf.rs b/src/net/bytes_buf.rs index 3929a860..1d928ffb 100644 --- a/src/net/bytes_buf.rs +++ b/src/net/bytes_buf.rs @@ -3,6 +3,8 @@ use std::collections::VecDeque; use bytes::BytesMut; +use crate::stream::ByteStream; + pub use bytes::Bytes; /// A circular buffer of bytes, internally represented as a list of Bytes @@ -119,6 +121,17 @@ impl BytesBuf { pub fn into_slices(self) -> VecDeque<Bytes> { self.buf } + + /// Return the entire buffer concatenated into a single big Bytes + pub fn into_bytes(mut self) -> Bytes { + self.take_all() + } + + /// Return the content as a stream of individual chunks + pub fn into_stream(self) -> ByteStream { + use futures::stream::StreamExt; + Box::pin(futures::stream::iter(self.buf).map(|x| Ok(x))) + } } impl Default for BytesBuf { diff --git a/src/net/stream.rs b/src/net/stream.rs index 88c3fed4..3ac6896d 100644 --- a/src/net/stream.rs +++ b/src/net/stream.rs @@ -200,3 +200,14 @@ pub fn asyncread_stream<R: AsyncRead + Send + Sync + 'static>(reader: R) -> Byte pub fn stream_asyncread(stream: ByteStream) -> impl AsyncRead + Send + Sync + 'static { tokio_util::io::StreamReader::new(stream) } + +/// Reads all of the content of a `ByteStream` into a BytesBuf +/// that contains everything +pub async fn read_stream_to_end(mut stream: ByteStream) -> Result<BytesBuf, std::io::Error> { + let mut buf = BytesBuf::new(); + while let Some(part) = stream.next().await { + buf.extend(part?); + } + + Ok(buf) +} |