如何监控Reqwest客户端上传进度。

Question

我正在使用下面的代码通过reqwest和tokio下载，并显示进度

pub async fn download_file(client: &Client, url: &str, path: &str) -> Result<(), String> {
    // Reqwest setup
    let res = client
        .get(url)
        .send()
        .await
        .or(Err(format!("Failed to GET from '{}'", &url)))?;
    let total_size = res
        .content_length()
        .ok_or(format!("Failed to get content length from '{}'", &url))?;

    // Indicatif setup
    let pb = ProgressBar::new(total_size);
    pb.set_style(ProgressStyle::default_bar()
        .template("{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})")
        .progress_chars("#>-"));
    pb.set_message(format!("Downloading {}", url));

    // download chunks
    let mut file = File::create(path).or(Err(format!("Failed to create file '{}'", path)))?;
    let mut downloaded: u64 = 0;
    let mut stream = res.bytes_stream();

    while let Some(item) = stream.next().await {
        let chunk = item.or(Err(format!("Error while downloading file")))?;
        file.write(&chunk)
            .or(Err(format!("Error while writing to file")))?;
        let new = min(downloaded + (chunk.len() as u64), total_size);
        downloaded = new;
        pb.set_position(new);
    }

    pb.finish_with_message(format!("Downloaded {} to {}", url, path));
    return Ok(());
}

通过while循环，我可以设置进度并查看进度条，类似于这里的示例https://github.com/mitsuhiko/indicatif

现在我正在尝试从上传中创建进度条，但是找不到监视reqwest客户端的方法，下面的代码是我的上传函数。

pub async fn upload_file(client: &Client, url: &str, path: &str) -> Result<(), String> {
    let f = File::open(path).expect("Unable to open file");

    let total_size = f.metadata().unwrap().len();

    // Indicatif setup
    let pb = ProgressBar::new(total_size);
    pb.set_style(ProgressStyle::default_bar()
        .template("{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})")
        .progress_chars("#>-"));
    pb.set_message(format!("Posting {}", url));


    let file = tokio::fs::File::open(path).await.unwrap();
    let stream = FramedRead::new(file, BytesCodec::new());


    let res=client
    .post(url)
    .body(Body::wrap_stream(stream))
    .send()
    .await;

    pb.finish_with_message(format!("Uploaded {} to {}", url, path));
    return Ok(());
}

上传可以完成，但没有百分比或其他指示进度的进度条。应该有状态监视器，就像下面这样。

    .post(url)
    .body(Body::wrap_stream(stream))
    .send()
    .monitorStatus(|stat|{
        pb.set_position(stat);
    }).....

你可以在这里查看工作代码：https://github.com/ozkanpakdil/rust-examples/blob/5f4965f2b086d07c8294352182639dc75232bb30/download_upload/src/download_file.rs#L43，只需取消注释这些测试并运行 cargo test

我的问题是，如何监视 reqwest 客户端以进行上传，并从中制作进度条？

- ozkanpakdil

Answer 1

您可以创建一个async_stream，并将输入的数据块作为yield返回以进行上传：

let file = tokio::fs::File::open(&input).await.unwrap();
let total_size = file.metadata().await.unwrap().len();
let input_ = input.to_string();
let output_ = output.to_string();
let mut reader_stream = ReaderStream::new(file);

let mut uploaded = HTTPSHandler::get_already_uploaded(output).await;
bar.set_length(total_size);

let async_stream = async_stream::stream! {
    while let Some(chunk) = reader_stream.next().await {
        if let Ok(chunk) = &chunk {
            let new = min(uploaded + (chunk.len() as u64), total_size);
            uploaded = new;
            bar.set_position(new);
            if(uploaded >= total_size){
                bar.finish_upload(&input_, &output_);
            }
        }
        yield chunk;
    }
};

然后，在构建Body时，只需将流包装起来：

let _ = reqwest::Client::new()
        .put(output)
        .header("content-type", "application/octet-stream")
        .header("Range", "bytes=".to_owned() + &uploaded.to_string() + "-")
        .header(
            reqwest::header::USER_AGENT,
            reqwest::header::HeaderValue::from_static(CLIENT_ID),
        )
        .body(reqwest::Body::wrap_stream(async_stream))
        .send()
        .await
        .unwrap();

顺便看看aim的实现，我在那里遇到了类似的问题！

- Mihai Galos