在 Cloud Storage 对象写入时推送数据

Rust 版 Cloud 客户端库中的 Cloud Storage 客户端通过从实现 StreamingSource 特征的类型中拉取载荷分块来写入对象。某些应用会在单独的任务或线程中生成数据，因此需要将载荷分块推送到上传。

本文档介绍了如何在 StreamingSource 实现中封装异步通道接收器，以便将数据推送到 Cloud Storage 对象写入。如果您不熟悉客户端库，请先参阅将 Cloud Storage 与 Rust 搭配使用。

准备工作

在运行本文档中的代码之前，请先完成以下前提条件：

• 创建或选择一个已启用结算功能的 Google Cloud 项目。
• 在项目中创建 Cloud Storage 存储桶。
• 配置应用默认凭证，并向您的主账号授予具有在存储桶中创建对象的权限的 Identity and Access Management (IAM) 角色，例如 Storage Object User (roles/storage.objectUser)。

依赖项

将 Cloud Storage 客户端库和本文档中使用的支持 crate 添加到 Cargo.toml 文件中：

cargo add google-cloud-storage anyhow bytes
cargo add tokio --features rt,sync

将队列接收器转换为流式传输来源

如需将生成数据的任务与上传载荷的任务解耦，请使用异步队列连接它们。本文档使用 Tokio mpsc 通道，但您可以使用与异步运行时集成的任何通道或队列。

如需将频道接收器调整为 StreamingSource 实现，请完成以下步骤：

1. 定义一个封装通道 Receiver 的结构体：

   use tokio::sync::mpsc::{self, Receiver};
   #[derive(Debug)]
   struct QueueSource(Receiver<bytes::Bytes>);

2. 为封装容器结构体实现 StreamingSource 特征，以便每次调用 next 时都等待接收器中的下一个数据块：

   use google_cloud_storage::streaming_source::StreamingSource;
   impl StreamingSource for QueueSource {
       type Error = std::convert::Infallible;
       async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
           self.0.recv().await.map(Ok)
       }
   }

将数据从队列流式传输到 Cloud Storage

实现 StreamingSource 特征后，您可以在后台任务中生成对象上传，并通过通道从生产者任务发送分块。由于通道接收器是单通道数据源，未实现 Seek，因此请使用 send_buffered 而不是 send_unbuffered 启动上传。

在接受格式为 projects/_/buckets/BUCKET_ID 和 object_name 的 bucket_name 的 async 函数中，完成以下步骤：

1. 初始化 Storage 客户端：

   use google_cloud_storage::client::Storage;
   let client = Storage::builder().build().await?;

2. 创建有界限的 mpsc 通道，以获取 sender 和 receiver 句柄：

   let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);

3. 将对象上传配置为以 QueueSource(receiver) 作为载荷来源。请勿等待 send_buffered future：

   let upload = client
       .write_object(bucket_name, object_name, QueueSource(receiver))
       .send_buffered();

4. 在生产者生成数据时，生成后台任务来驱动上传：

   let task = tokio::spawn(upload);

5. 在提供方任务中，将数据块推送到 sender 句柄中：

   for _ in 0..1000 {
       let line = "I will not write funny examples in class\n";
       sender
           .send(bytes::Bytes::from_static(line.as_bytes()))
           .await?;
   }

6. 发送所有数据后，丢弃 sender 句柄以关闭通道并发出对象流结束的信号：

   drop(sender);

7. 等待后台任务句柄，以验证任务和上传请求是否都已成功，并检索上传的 Object 元数据：

   let object = task.await??;
   println!("object successfully uploaded {object:?}");

注意：如果上传在所有文件块发送之前失败，则后台任务会退出并丢弃接收器，从而导致 sender.send 返回关闭通道错误。在生产代码中，如果 sender.send 失败，请等待后台任务句柄检查底层 Cloud Storage 错误。

清理

为避免因本文档中使用的资源导致您的 Google Cloud 账号产生费用，请在完成后删除您上传的测试对象。

完整代码

以下示例将上述所有步骤合并为一个完整的函数：

use google_cloud_storage::streaming_source::StreamingSource;
use tokio::sync::mpsc::{self, Receiver};
#[derive(Debug)]
struct QueueSource(Receiver<bytes::Bytes>);
impl StreamingSource for QueueSource {
    type Error = std::convert::Infallible;
    async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
        self.0.recv().await.map(Ok)
    }
}

pub async fn queue(bucket_name: &str, object_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::client::Storage;
    let client = Storage::builder().build().await?;

    let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);
    let upload = client
        .write_object(bucket_name, object_name, QueueSource(receiver))
        .send_buffered();
    let task = tokio::spawn(upload);

    for _ in 0..1000 {
        let line = "I will not write funny examples in class\n";
        sender
            .send(bytes::Bytes::from_static(line.as_bytes()))
            .await?;
    }
    drop(sender);
    let object = task.await??;
    println!("object successfully uploaded {object:?}");

    Ok(())
}