Creating a Stream<Item=Result<Bytes>> from Tokio File

1

I'm attempting to create a streaming reqwest::Body from a file on my filesystem and I'm using Tokio, so it's async.

I'm getting reference issues as I have to borrow the buffer as mutable in my core stream loop and immutable in the body of the loop to read out of it.

Example code:

async fn file(path: PathBuf) -> Result<impl Stream<Item=Result<Bytes, anyhow::Error>>, anyhow::Error> {
    let stream = try_stream! {
        let mut f = tokio::fs::File::open(file).await.unwrap();

        let mut buf = BytesMut::with_capacity(8 * 1024);

        while let bytes_read = f.read(buf.as_mut()).await? {
            if bytes_read == 0 { break }

            yield Bytes::from(&buf[..bytes_read]);
        }
    }

    Ok(stream)
}

I don't see a way to make buf non-mutable, as the read function needs a mutable reference, and I must be able to read from it in order to create a Bytes from it.

How should I go about doing this? Are there any examples out there for this? This seems like a common thing to do and I'm not seeing a path forward. I suppose I could wrap it all in a Mutex or a RwLock but that feels like overkill here.

2

Figured it out :duck:

The call to BytesMut::from was the problem. Bytes and BytesMut are efficient, so the underlying Bytes value that is created possibly contains (among other things probably):duck:

  1. a reference to the data
  2. or the owned data

Essentially it's kind of like this (except not an enum):

enum Bytes<'_> {
    Reference(&'_ u8),
    Owned(Vec<u8>),
}

So my call to Bytes::from was creating the reference variant:

yield Bytes::from(&buf[..bytes_read]);

Simply creating a Vec with this fixed the problem by removal of the reference:

yield Bytes::from(buf[..bytes_read].to_vec());
3

doesn't Bytes requires 'static if you want to construct from a slice? Bytes simply cannot borrow/reuse the buffer, you must allocate a new buffer for each chunk of Bytes you yield. in other words, move let mut buf inside the loop.

also, your use of BytesMut::as_mut() to read the file into is incorrect. like a Vec here, when you create a BytesMut with capacity of 8K, it's still empty, it has the desired capacity(), but the len() is 0.

you don't need the BytesMut here, just use a Vec is ok, Bytes can be constructed from Vec<u8> directly:

loop {
    let mut buf = vec![0u8; CHUNK_SIZE];
    let nbytes = f.read(&mut buf).await?:
    if nbytes == 0 {
        break;
    }
    buf.truncate(nbytes);
    yield Bytes::from(buf)
}

an allocation per chunk is not ideal, but it is what the Stream<Item = Bytes> api requires. you don't have to stream the file if your file size is not extremely large, and you have enough memory budget. you trade the memory usage for simpler code and a slightly better performance.

4

are you talking the Bytes type from the bytes crate? if so, it doesn't have a "reference variant" (except for 'static slices). you simply cannot create a Bytes out of a borrow that is not 'static.