Mendorong data pada penulisan objek Cloud Storage

Klien Cloud Storage di Library Klien Cloud untuk Rust menulis objek dengan menarik potongan payload dari jenis yang mengimplementasikan trait StreamingSource. Beberapa aplikasi menghasilkan data dalam tugas atau thread terpisah dan perlu mengirimkan bagian payload ke upload.

Dokumen ini menunjukkan cara menggabungkan penerima saluran asinkron dalam implementasi StreamingSource sehingga Anda dapat mengirim data ke penulisan objek Cloud Storage. Jika Anda baru menggunakan library klien, lihat Menggunakan Cloud Storage dengan Rust terlebih dahulu.

Sebelum memulai

Sebelum menjalankan kode dalam dokumen ini, selesaikan prasyarat berikut:

• Buat atau pilih projectGoogle Cloud dengan penagihan diaktifkan.
• Buat bucket Cloud Storage di project Anda.
• Konfigurasikan Kredensial Default Aplikasi dan berikan peran Identity and Access Management (IAM) kepada akun utama Anda dengan izin untuk membuat objek di bucket, seperti Storage Object User (roles/storage.objectUser).

Dependensi

Tambahkan library klien Cloud Storage dan crate pendukung yang digunakan dalam dokumen ini ke file Cargo.toml Anda:

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

Mengonversi penerima antrean menjadi sumber streaming

Untuk memisahkan tugas yang menghasilkan data dari tugas yang mengupload payload, hubungkan keduanya dengan antrean asinkron. Dokumen ini menggunakan saluran mpsc Tokio, meskipun Anda dapat menggunakan saluran atau antrean apa pun yang terintegrasi dengan runtime asinkron Anda.

Untuk menyesuaikan penerima saluran ke dalam implementasi StreamingSource, selesaikan langkah-langkah berikut:

1. Tentukan struct yang menggabungkan saluran Receiver:

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

2. Terapkan trait StreamingSource untuk struct wrapper sehingga setiap panggilan ke next menunggu potongan berikutnya dari penerima:

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

Melakukan streaming data dari antrean ke Cloud Storage

Setelah mengimplementasikan trait StreamingSource, Anda dapat memunculkan upload objek dalam tugas latar belakang dan mengirim potongan melalui saluran dari tugas produsen Anda. Karena penerima saluran adalah sumber data sekali-proses yang tidak mengimplementasikan Seek, mulai upload menggunakan send_buffered, bukan send_unbuffered.

Di dalam fungsi async yang menerima bucket_name dalam format projects/_/buckets/BUCKET_ID dan object_name, selesaikan langkah-langkah berikut:

1. Lakukan inisialisasi klien Storage:

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

2. Buat saluran mpsc terbatas untuk mendapatkan handle sender dan receiver:

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

3. Konfigurasikan upload objek dengan QueueSource(receiver) sebagai sumber payload. Jangan menunggu future send_buffered:

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

4. Buat tugas latar belakang untuk mendorong upload saat produsen Anda membuat data:

   let task = tokio::spawn(upload);

5. Dalam tugas produsen, kirim potongan data ke handle 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. Setelah mengirim semua data, hapus handle sender untuk menutup saluran dan memberi sinyal akhir aliran objek:

   drop(sender);

7. Tunggu pengendali tugas latar belakang untuk memverifikasi bahwa tugas dan permintaan upload berhasil, lalu ambil metadata Object yang diupload:

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

Catatan: Jika upload gagal sebelum semua potongan dikirim, tugas latar belakang akan keluar dan menghapus penerima, sehingga sender.send menampilkan error saluran tertutup. Dalam kode produksi, jika sender.send gagal, tunggu handle tugas latar belakang untuk memeriksa error Cloud Storage yang mendasarinya.

Pembersihan

Agar akun Google Cloud Anda tidak dikenai biaya untuk resource yang digunakan dalam dokumen ini, hapus objek pengujian yang Anda upload setelah selesai.

Kode lengkap

Contoh berikut menggabungkan semua langkah sebelumnya menjadi fungsi yang lengkap:

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