stream - 如何将 futures::Stream 写入磁盘而不先将其完全存储在内存中？

Question

这里有一个使用 Rusoto S3 下载文件的示例: How to save a file downloaded from S3 with Rusoto to my hard drive?

问题是它看起来像是将整个文件下载到内存中，然后将其写入磁盘，因为它使用了 write_all方法采用字节数组，而不是流。如何使用 StreamingBody , 它实现了 futures::Stream将文件流式传输到磁盘？

Answer 1

由于 StreamingBody 实现了 Stream<Item = Vec<u8>, Error = Error> ，我们可以构建一个表示它的 MCVE :

extern crate futures; // 0.1.25

use futures::{prelude::*, stream};

type Error = Box<std::error::Error>;

fn streaming_body() -> impl Stream<Item = Vec<u8>, Error = Error> {
    const DUMMY_DATA: &[&[u8]] = &[b"0123", b"4567", b"89AB", b"CDEF"];
    let iter_of_owned_bytes = DUMMY_DATA.iter().map(|&b| b.to_owned());
    stream::iter_ok(iter_of_owned_bytes)
}

然后我们可以以某种方式获得“流式主体”并使用 Stream::for_each 来处理 Stream 中的每个元素。在这里，我们只调用 write_all 和一些提供的输出位置:

use std::{fs::File, io::Write};

fn save_to_disk(mut file: impl Write) -> impl Future<Item = (), Error = Error> {
    streaming_body().for_each(move |chunk| file.write_all(&chunk).map_err(Into::into))
}

然后我们可以编写一个小的测试主程序:

fn main() {
    let mut file = Vec::new();

    {
        let fut = save_to_disk(&mut file);
        fut.wait().expect("Could not drive future");
    }

    assert_eq!(file, b"0123456789ABCDEF");
}

关于这种天真的实现质量的重要说明:

1. 对 write_all 的调用可能会阻塞，您不应该在异步程序中这样做。最好将阻塞工作交给线程池。

2. Future::wait 的使用强制线程阻塞，直到 future 完成，这对测试非常有用，但对于您的实际用例可能不正确。

另见:

• What is the best approach to encapsulate blocking I/O in future-rs?
• How do I synchronously return a value calculated in an asynchronous Future in stable Rust?