how to create a stream from an in-memory buffer

Question

I am reading from STDIN into a big buffer (BytesMut) in memory, I am using something like this:

use anyhow::Result;
use bytes::{BufMut, BytesMut};
use futures::stream::TryStreamExt;
use tokio::io::stdin;
use tokio_util::codec::{BytesCodec, FramedRead};

#[tokio::main]
async fn main() -> Result<()> {
    let mut buffer = BytesMut::with_capacity(1024 * 1024 * 512);
    let mut input_stream = FramedRead::new(stdin(), BytesCodec::new());
    while let Some(bytes) = input_stream.try_next().await? {
        buffer.put(bytes);
    }
    println!("buf: {}", buffer.len());
    Ok(())
}

buffer now contains the full input, but how can I create a stream from it?

I want a stream because I would like to POST in streams instead of just trying to push the whole buffer by using the reqwest create using something like:

 let client = reqwest::Client::new();
 let body = reqwest::Body::wrap_stream(stream);
 client.post().body(body)

In my naive attempt, I thought that passing the buffer as an argument for FrameRead it could work, something like:

let mut output_stream = FramedRead::new(buffer, BytesCodec::new());

But that gives the error:

the trait bound `bytes::BytesMut: tokio::io::AsyncRead` is not satisfied

Then I found in the AsyncRead docs that there is the method poll_read_buf that may help to convert the buffer into a stream, but here is where I am stuck trying to find how to implement or create the stream from an in-memory buffer.

My current try is with the create async-stream with something like this:

let stream = async_stream::stream! {
    while !buffer.is_empty() {
        let out = if buffer.len() > 8192 {
            buffer.split_to(8192)
        } else {
            buffer.split_to(buffer.len())
        };
        yield out.freeze();
    }
};

I think that I am getting close but that returns:

 --> src/main.rs:32:46
   |
32 |     let body = Body::wrap_stream(Bytes::from(stream));
   |                                              ^^^^^^ the trait `std::convert::From>` is not implemented for `bytes::Bytes`
   |
   = help: the following implementations were found:
             >
             >
             >
             >
           and 4 others
   = note: required by `std::convert::From::from`

error[E0277]: the trait bound `bytes::Bytes: tokio::stream::Stream` is not satisfied
  --> src/main.rs:32:34
   |
32 |     let body = Body::wrap_stream(Bytes::from(stream));
   |                                  ^^^^^^^^^^^^^^^^^^^ the trait `tokio::stream::Stream` is not implemented for `bytes::Bytes`

How from my big in-memory buffer I could now create chunks and build/return a Stream?

Any ideas?

how to create a stream from an in-memory buffer

0 Answers