在 Cloud Storage 物件寫入時推送資料

Rust 適用的 Cloud 用戶端程式庫中的 Cloud Storage 用戶端，會從實作 StreamingSource 特徵的類型提取酬載區塊，藉此寫入物件。部分應用程式會在獨立的工作或執行緒中產生資料，因此需要將酬載區塊推送至上傳作業。

本文說明如何將非同步管道接收器包裝在 StreamingSource 實作中，以便將資料推送至 Cloud Storage 物件寫入。如果您是第一次使用用戶端程式庫，請先參閱「搭配使用 Cloud Storage 與 Rust」。

事前準備

執行本文件中的程式碼前，請先完成下列必要條件：

• 建立或選取已啟用計費功能的 Google Cloud 專案。
• 在專案中建立 Cloud Storage bucket。
• 設定應用程式預設憑證，並將具備值區物件建立權限的 Identity and Access Management (IAM) 角色 (例如 Storage 物件使用者 (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 未來：

   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(())
}