如何将Rust的AsyncRead对象转换为ByteStream并上传至S3？

解决方案 #

使用tokio_util::io::ReaderStream直接将AsyncRead转换为ByteStream所需的Stream类型，这是最直接的适配方式：

步骤1：添加依赖 #

确保Cargo.toml中包含tokio-util（启用io特性）：

tokio-util = { version = "0.7", features = ["io"] }

步骤2：修改上传代码 #

use aws_sdk_s3::types::ByteStream;
use tokio_util::io::ReaderStream;
use std::error::Error;
use tokio::io::AsyncRead;

async fn upload(&self, key: String, data: &mut impl AsyncRead + Unpin) -> Result<(), Box<dyn Error>> {
    // 将AsyncRead转换为符合ByteStream要求的Stream
    let byte_stream = ByteStream::from(ReaderStream::new(data));

    self.s3
        .put_object()
        .key(key)
        .body(byte_stream)
        .send()
        .await?;

    Ok(())
}

处理Unpin约束 #

如果你的AsyncRead对象没有实现Unpin（比如某些自定义异步读取器），可以用tokio::pin!宏固定它：

use tokio::pin;

async fn upload(&self, key: String, mut data: impl AsyncRead) -> Result<(), Box<dyn Error>> {
    pin!(data); // 固定对象以满足Unpin约束
    let byte_stream = ByteStream::from(ReaderStream::new(&mut data));

    self.s3
        .put_object()
        .key(key)
        .body(byte_stream)
        .send()
        .await?;

    Ok(())
}

错误原因分析 #

你之前遇到的问题根源：

1. 特征约束未满足：FramedRead对底层读取器的约束更严格，且未正确处理&mut AsyncRead的引用类型；而ReaderStream原生支持&mut AsyncRead（仅需满足Unpin）。
2. FramedRead适配冗余：FramedRead是用于编解码场景的Stream，本身不是迭代器，且需要额外处理BytesMut转Bytes的步骤；ReaderStream直接生成ByteStream所需的Stream<Item = Result<Bytes, std::io::Error>>，无需多余编解码操作。

内容的提问来源于stack exchange，提问作者amit