Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Cloud Storage オブジェクトの書き込み時にデータを push する

Rust 用 Cloud クライアントライブラリの Cloud Storage クライアントは、StreamingSource 特性を実装する型からペイロードチャンクを pull してオブジェクトを書き込みます。一部のアプリケーションでは、別のタスクまたはスレッドでデータを生成し、代わりにペイロードチャンクをアップロードに push する必要があります。

このドキュメントでは、非同期チャネルレシーバを StreamingSource 実装でラップして、Cloud Storage オブジェクトの書き込みにデータを push する方法について説明します。クライアントライブラリを初めて使用する場合は、まず Rust で Cloud Storage を使用するをご覧ください。

始める前に

このドキュメントのコードを実行する前に、次の前提条件を満たしてください。

課金が有効になっている Google Cloud プロジェクトを作成または選択します。
プロジェクトで Cloud Storage バケットを作成します。
アプリケーションのデフォルト認証情報を構成し、バケットにオブジェクトを作成する権限（Storage オブジェクトユーザー（roles/storage.objectUser）など）を持つ Identity and Access Management（IAM）ロールをプリンシパルに付与します。

依存関係

このドキュメントで使用されている Cloud Storage クライアントライブラリとサポートクレートを Cargo.toml ファイルに追加します。

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

キューレシーバーをストリーミングソースに変換する

データを生成するタスクとペイロードをアップロードするタスクを切り離すには、非同期キューで接続します。このドキュメントでは、Tokio の mpsc チャネルを使用しますが、非同期ランタイムと統合する任意のチャネルまたはキューを使用できます。

チャンネルレシーバーを StreamingSource 実装に適合させるには、次の手順を行います。

チャネル Receiver をラップする構造体を定義します。

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

ラッパー構造体に 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_unbuffered ではなく send_buffered を使用してアップロードを開始します。

projects/_/buckets/BUCKET_ID 形式と object_name 形式の bucket_name を受け入れる async 関数内で、次の手順を完了します。

Storage クライアントを初期化します。

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

sender ハンドルと receiver ハンドルを取得するために、境界付き mpsc チャンネルを作成します。
```
let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);
```
ペイロードソースとして QueueSource(receiver) を使用してオブジェクトのアップロードを構成します。send_buffered の future はまだ待機しないでください。
```
let upload = client
    .write_object(bucket_name, object_name, QueueSource(receiver))
    .send_buffered();
```
プロデューサーがデータを生成している間に、アップロードを駆動するバックグラウンドタスクを生成します。
```
let task = tokio::spawn(upload);
```

プロデューサータスクで、データチャンクを sender ハンドルに push します。

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?;
}

すべてのデータを送信したら、sender ハンドルをドロップしてチャネルを閉じ、オブジェクトストリームの終了を通知します。
```
drop(sender);
```
バックグラウンドタスクハンドルを待機して、タスクとアップロードリクエストの両方が成功したことを確認し、アップロードされた Object メタデータを取得します。
```
let object = task.await??;
println!("object successfully uploaded {object:?}");
```

クリーンアップ

このドキュメントで使用したリソースについて Google Cloud アカウントに課金されないようにするには、アップロードしたテストオブジェクトを削除します。

コードを補完する

次の例では、上記のすべてのステップを 1 つの関数にまとめています。

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