rust - 是否可以从文件创建流而不是将文件内容加载到内存中？

Question

我目前正在使用rusoto_s3 lib 将文件上传到 S3。我发现的所有示例都执行相同的操作:打开文件，将文件的完整内容读入内存( Vec<u8> )，然后将 Vec 转换为 ByteStream 。 (实现 From<Vec<u8>> )。这是一个代码示例:

fn upload_file(&self, file_path: &Path) -> FileResult<PutObjectOutput> {
    let mut file = File::open(file_path)?;
    let mut file_data: Vec<u8> = vec![];
    file.read_to_end(&mut file_data)?;

    let client = S3Client::new(Region::UsEast1);
    let mut request = PutObjectRequest::default();
    request.body = Some(file_data.into());

    Ok(client.put_object(request).sync()?)
}

这对于小文件来说可能是可以接受的，但是(我假设)一旦您尝试上传大小大于可用堆内存的文件，这种技术就会崩溃。

另一种创建 ByteStream 的方法来自 using this initializer它接受一个实现 Stream 的对象特征。我假设File会实现这个特性，但事实似乎并非如此。

我的问题:

是否有某种类型可以从 File 构造出来它实现了 Stream ？是制作我自己的包装 File 的元组结构的正确解决方案并实现Stream本身，这个实现是微不足道的吗？是否有其他我没有看到的解决方案，或者我只是误解了上面代码中的内存分配方式？

Answer 1

Is there some type which can be constructed from a File which implements Stream?

不幸的是，不。目前，std、futures 或 tokio 中没有任何内置功能可以直接执行此操作。

由于 Stream 项目的“分离”性质，这样的实现必须为传入数据的每个切片分配一个新的拥有的缓冲区并将其移交给调用者。那效率不会很高。直到 Rust 语言拥有泛型关联类型 (GAT)(希望在明年)，我们才能满意地解决这个问题。查看this futures-rs ticket和 Niko's async interview #2了解更多详情。

话虽如此，目前在某些用例中，底层 IO 之上的 Stream 外观是可取的并且足够好。

Is the correct solution to make my own tuple struct which wraps File and implements Stream itself, and is this implementation trivial?

对于rusoto依赖的futures-0.1，有几种方法可以实现:

• 为包装 Read 的结构实现 Stream 特征
• 利用 futures 实用函数，例如 futures::stream::poll_fn
• tokio-codec-0.1 有一个出色的 FramedRead，它已经实现了 Stream

第三个肯定是最简单的:

use futures::stream::Stream;  // futures = "0.1.29"
use rusoto_core::{ByteStream, Region};  // rusoto_core = "0.42.0"
use rusoto_s3::{PutObjectOutput, PutObjectRequest, S3Client, S3};  // rusoto_s3 = "0.42.0"
use std::{error::Error, fs::File, path::Path};
use tokio_codec::{BytesCodec, FramedRead};  // tokio-codec = "0.1.1"
use tokio_io::io::AllowStdIo;  // tokio-io = "0.1.12"

fn upload_file(file_path: &Path) -> Result<PutObjectOutput, Box<dyn Error>> {
    let file = File::open(file_path)?;
    let aio = AllowStdIo::new(file);
    let stream = FramedRead::new(aio, BytesCodec::new()).map(|bs| bs.freeze());

    let client = S3Client::new(Region::UsEast1);
    let mut request = PutObjectRequest::default();
    request.body = Some(ByteStream::new(stream));

    Ok(client.put_object(request).sync()?)
}