用rusoto流上传到s3

Question

如何使用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");
}

我收到以下错误：

errorE0277:属性绑定std::io::BufReader<std::fs::File>: futures::stream::Stream不满意-> src/bin/example.rs: 18 :20 \x#*

Answer 1

好吧。系好安全带，这是一件有趣的事。

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

显然，BufReader没有实现这个特性，因为它比未来和流早了很长一段时间。也没有任何简单的转换到Stream<Item = Bytes>，您可以使用它来隐式地转换为此。

第一个(注释)示例之所以工作，是因为String::into_bytes().into()将遵循类型传递链：String -> Vec<u8> -> ByteStream，这要归功于From<Vec<u8>>在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

Answer 2

下面是一个即将推出的Rusoto异步等待语法的版本(对于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);
}

它本质上是从集成测试套件本身借来的，在那里您可以找到上传版本的片段。