8

如何使用rusoto将文件上传到 s3 ，而不将文件内容读取到内存（流式传输）？

使用此代码：

use std::fs::File;
use std::io::BufReader;

use rusoto_core::Region;
use rusoto_s3::{PutObjectRequest, S3, S3Client, StreamingBody};

fn main() {
    let file = File::open("input.txt").unwrap();
    let mut reader = BufReader::new(file);

    let s3_client = S3Client::new(Region::UsEast1);
    let result = s3_client.put_object(PutObjectRequest {
        bucket: String::from("example_bucket"),
        key: "example_filename".to_string(),
//        this works:
//      body: Some("example string".to_owned().into_bytes().into()),
//        this doesn't:
        body: Some(StreamingBody::new(reader)),
        ..Default::default()
    }).sync().expect("could not upload");
}

我收到以下错误：

error[E0277]: the trait bound `std::io::BufReader<std::fs::File>: futures::stream::Stream` is not satisfied
  --> src/bin/example.rs:18:20
   |
18 |         body: Some(StreamingBody::new(reader)),
   |                    ^^^^^^^^^^^^^^^^^^ the trait `futures::stream::Stream` is not implemented for `std::io::BufReader<std::fs::File>`
   |
   = note: required by `rusoto_core::stream::ByteStream::new`
4

2 回答 2

20

好的。把自己绑起来，这很有趣。

StreamingBody是 的别名ByteStream，它本身采用参数类型S: Stream<Item = Bytes, Error = Error> + Send + 'static。简而言之，它需要是一个字节流。

BufReader显然，它没有实现这个特性，因为它比期货和流早了很长时间。也没有简单的转换Stream<Item = Bytes>可以用来隐式转换成 this。

第一个（注释）示例有效的原因是因为String::into_bytes().into()将遵循类型转换链：String-> Vec<u8>->ByteStream感谢From<Vec<u8>>on的实现ByteStream。

现在我们知道为什么这不起作用，我们可以修复它。有快速的方法，然后有正确的方法。我会告诉你两个。

快速的方法

快速（但不是最佳）的方法是简单地调用File::read_to_end(). 这将填充一个Vec<u8>，然后您可以像以前一样使用它：

 let mut buf:Vec<u8> = vec![];
 file.read_to_end(&mut buf)?;
 // buf now contains the entire file

这是低效且次优的，原因有两个：

  • read_to_end()是阻塞调用。根据您从哪里读取文件，这个阻塞时间可能被证明是不合理的
  • 您需要拥有比文件中的字节更多的可用 RAM（+ 64 或 128 位用于Vec定义 + 一些我们并不真正关心的额外）

好办法

好方法将您的文件转换为实现AsyncRead. 由此，我们可以形成一个Stream.

既然你已经有了std::fs::File，我们先把它转换成tokio::fs::File。这实现AsyncRead了，这对以后非常重要：

let tokio_file = tokio::fs::File::from_std(file);

因此，遗憾的是，我们需要做一些管道工作才能将其放入Stream. 多个板条箱已经实现了它；从头开始的方法如下：

use tokio_util::codec;
let byte_stream = codec::FramedRead::new(tokio_file, codec::BytesCodec::new())
   .map(|r| r.as_ref().to_vec());

byte_stream是一个实例，tokio_util::codec::FramedRead它基于我们的解码器使用特定项目实现。Stream由于我们的解码器是BytesCodec，因此您的流是Stream<Item = BytesMut>。

由于操场不知道rusoto_core，我无法向您展示完整的流程。但是，我可以向您展示您可以生成一个Stream<Item = Vec<u8>, Error = io::Error>，这是关键：https ://play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=38e4ae8be0d70abd134b5331d6bf4133

于 2019-09-05T19:59:16.407 回答
6

这是一个带有即将推出的 Rusoto async-await 语法的版本（对于 getObject 虽然应该可以直接调整上传）......可能在 Rusoto 0.4.3 中供公众使用：

https://github.com/brainstorm/rusoto-s3-async-await

即：

pub async fn bucket_obj_bytes(client: S3Client, bucket: String, _prefix: String, object: String) {
    let get_req = GetObjectRequest {
        bucket,
        key: object,
        ..Default::default()
    };

    let result = client
        .get_object(get_req)
        .await
        .expect("Couldn't GET object");
    println!("get object result: {:#?}", result);

    let stream = result.body.unwrap();
    let body = stream.map_ok(|b| BytesMut::from(&b[..])).try_concat().await.unwrap();

    assert!(body.len() > 0);
    dbg!(body);
}

这本质上是从集成测试套件本身借来的，您也可以在其中找到上传版本的片段。

于 2020-01-23T17:35:01.760 回答