Async http PUT in loop with reqwest

[mihaigalos]

In Rust, using reqwest and tokio, I'm trying to put a file to a http server (nginx, hosted locally).
I would like to have a loop and update a progress bar with the amount of bytes transfered:

pub async fn put(input: &str, output: &str, _: &WrappedBar) {
    let file = tokio::fs::File::open(input).await.unwrap();
    let stream = tokio_util::codec::FramedRead::new(file, tokio_util::codec::BytesCodec::new());
    let body = reqwest::Body::wrap_stream(stream);

    let _ = reqwest::Client::new()
        .put(output)
        .header("content-type", "application/octet-stream")
        .body(body)
        .send()
        .await
        .unwrap();

    loop {
        let uploaded = file.seek(std::io::SeekFrom::Current(0)).await.unwrap();

        // use the client.body(body).send().await.unwrap() here instead of outside the loop
        // update the progress bar

        if uploaded == file.metadata().await.unwrap().len() {
            break;
        }
    }
}

Cargo.toml:

reqwest = { version = "0.11.3", features = ["stream"] }
tokio = { version = "1.14.0", features = ["full"] }
tokio-util = "0.6.9"

file is moved when constructing the stream, so how can I reuse it here? Is the approach even correct?

[Darksonn]

This is not possible. You will need to set up some other way to transmit this data. For example, you can create your own body using the async-stream crate which behaves like your codec, except that it also transmits the number of bytes read to a watch channel.

[Darksonn]

You can avoid consuming the chunk in this manner:

if let Ok(chunk) = &mut chunk {
    total += chunk.len();
}

[mihaigalos]

I think we're almost there, but not quite:
Tried to borrow as &mut:

let async_stream = async_stream::stream! {
    while let Some(chunk) = reader_stream.next().await {
        if let Ok(chunk) = &mut chunk {
             total += chunk.len();
        }
        yield chunk;
    }
};

Result: Error: cannot borrow chunk as mutable, as it is not declared as mutable.

So then tried to make chunk mutable:

let async_stream = async_stream::stream! {
    while let mut Some(chunk) = reader_stream.next().await {
        if let Ok(chunk) = &mut chunk {
             total += chunk.len();
        }
        yield chunk;
    }
};

Result:

error: expected `=`
  --> src/https.rs:48:28
   |
48 |           let async_stream = async_stream::stream! {
   |  ____________________________^
49 | |             while let mut Some(chunk) = reader_stream.next().await {
50 | |                 if let Ok(chunk) = &mut chunk {
51 | |                     total += chunk.len();
...  |
54 | |             }
55 | |         };
   | |_________^
   |
   = note: this error originates in the macro `$crate::async_stream_impl::stream_inner` (in Nightly builds, run with -Z macro-backtrace for more info)

[Darksonn]

The mut goes on the chunk and not on the Some.

[Darksonn]

Also, I realize that an immutable reference would work too - not sure why I put a mutable one.

[mihaigalos]

Oh yeah, true! Thank you very much for your precious help.
It now works.
Here's the current form for future reference:

pub async fn put(input: &str, output: &str, _: &WrappedBar) {
    let file = tokio::fs::File::open(&input).await.unwrap();
    let mut reader_stream = ReaderStream::new(file);

    let mut total = 0;
    let async_stream = async_stream::stream! {
        while let Some(chunk) = reader_stream.next().await {
            if let Ok(chunk) = &chunk {
                total += chunk.len();
            }
            yield chunk;
        }
    };

    let _ = reqwest::Client::new()
        .put(output)
        .header("content-type", "application/octet-stream")
        .body(reqwest::Body::wrap_stream(async_stream))
        .send()
        .await
        .unwrap();
}