rust - 如何在下载过程中解压缩和解压 tar.gz 存档？

Question

我需要在下载过程中解压和解压缩大的 .tar.gz 文件(例如 ~5Gb)，而不需要将存档文件保存在磁盘上。 我使用 reqwest crate 下载文件，flate2 crate 解压，tar crate 解包。我尝试用 tar.gz 格式来做。但是可以使用 zip 和 tar.bz2 格式。 (哪个更容易使用？) 看起来我成功地实现了这一点，但没想到解包以错误结束:

thread 'main' panicked at 'Cannot unpack archive: Custom { kind: UnexpectedEof, error: TarError { desc: "failed to unpack `/home/ruut/Projects/GreatWar/launcher/gamedata/gamedata-master/.vscode/settings.json`", io: Custom { kind: UnexpectedEof, error: TarError { desc: "failed to unpack `gamedata-master/.vscode/settings.json` into `/home/ruut/Projects/GreatWar/launcher/gamedata/gamedata-master/.vscode/settings.json`", io: Kind(UnexpectedEof) } } } }', /home/ruut/Projects/GreatWar/launcher/src/gitlab.rs:87:38
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

我的代码:

let full_url = format!("{}/{}/{}", HOST, repo_info.url, repo_info.download_url);
let mut response;

match self.client.get(&full_url).send().await {
  Ok(res) => response = res,
  Err(error) => {
    return Err(Error::new(ErrorKind::InvalidData, error));
  }
};

if response.status() == reqwest::StatusCode::OK {
  let mut stream = response.bytes_stream();

  while let Some(item) = stream.next().await {
    let chunk = item
      .or(Err(format!("Error while downloading file")))
      .unwrap();

    let b: &[u8] = &chunk.to_vec();
    let gz = GzDecoder::new(b);
    let mut archive = Archive::new(gz);

    archive.unpack("./gamedata").expect("Cannot unpack archive");
  }
}

archive.unpack 在第一次获取 block 后抛出错误。
我做错了什么？

Answer 1

kmdreko 的评论解释了为什么您的代码失败 - .next() 仅返回第一个 block ，您必须将所有 block 提供给 gzip 阅读器。另一个答案显示了如何使用阻塞 reqwest API 来做到这一点。

如果您想继续使用非阻塞 API，那么您可以在单独的线程中启动解码器并通过 channel 为其提供数据。例如，您可以使用 flume同时支持同步和异步接口(interface)的 channel 。您还需要将 channel 转换为 Read 的内容，如 GzDecoder 所期望的那样。例如(编译，但未经测试):

use std::io::{self, Read};

use flate2::read::GzDecoder;
use futures_lite::StreamExt;
use tar::Archive;

async fn download() -> io::Result<()> {
    let client = reqwest::Client::new();

    let full_url = "...";
    let response;

    match client.get(full_url).send().await {
        Ok(res) => response = res,
        Err(error) => {
            return Err(io::Error::new(io::ErrorKind::InvalidData, error));
        }
    };

    let (tx, rx) = flume::bounded(0);

    let decoder_thread = std::thread::spawn(move || {
        let input = ChannelRead::new(rx);
        let gz = GzDecoder::new(input);
        let mut archive = Archive::new(gz);
        archive.unpack("./gamedata").unwrap();
    });

    if response.status() == reqwest::StatusCode::OK {
        let mut stream = response.bytes_stream();

        while let Some(item) = stream.next().await {
            let chunk = item
                .or(Err(format!("Error while downloading file")))
                .unwrap();
            tx.send_async(chunk.to_vec()).await.unwrap();
        }
        drop(tx); // close the channel to signal EOF
    }

    tokio::task::spawn_blocking(|| decoder_thread.join())
        .await
        .unwrap()
        .unwrap();

    Ok(())
}

// Wrap a channel into something that impls `io::Read`
struct ChannelRead {
    rx: flume::Receiver<Vec<u8>>,
    current: io::Cursor<Vec<u8>>,
}

impl ChannelRead {
    fn new(rx: flume::Receiver<Vec<u8>>) -> ChannelRead {
        ChannelRead {
            rx,
            current: io::Cursor::new(vec![]),
        }
    }
}

impl Read for ChannelRead {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.current.position() == self.current.get_ref().len() as u64 {
            // We've exhausted the previous chunk, get a new one.
            if let Ok(vec) = self.rx.recv() {
                self.current = io::Cursor::new(vec);
            }
            // If recv() "fails", it means the sender closed its part of
            // the channel, which means EOF. Propagate EOF by allowing
            // a read from the exhausted cursor.
        }
        self.current.read(buf)
    }
}