Daten bei Schreibvorgängen in Cloud Storage-Objekte übertragen

Der Cloud Storage-Client in den Cloud-Clientbibliotheken für Rust schreibt Objekte, indem er Nutzlastblöcke aus einem Typ abruft, der das StreamingSource-Trait implementiert. Einige Anwendungen generieren Daten in einer separaten Aufgabe oder einem separaten Thread und müssen stattdessen Nutzlastblöcke an den Upload senden.

In diesem Dokument wird gezeigt, wie Sie einen asynchronen Kanalempfänger in eine StreamingSource-Implementierung einbinden, um Daten in einen Cloud Storage-Objektschreibvorgang zu übertragen. Wenn Sie die Clientbibliothek noch nicht kennen, lesen Sie zuerst Cloud Storage mit Rust verwenden.

Hinweis

Bevor Sie den Code in diesem Dokument ausführen, müssen die folgenden Voraussetzungen erfüllt sein:

Abhängigkeiten

Fügen Sie die Cloud Storage-Clientbibliothek und die in diesem Dokument verwendeten unterstützenden Crates der Datei Cargo.toml hinzu:

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

Warteschlangenempfänger in eine Streamingquelle umwandeln

Wenn Sie die Aufgabe, die Daten erzeugt, von der Aufgabe, die die Nutzlast hochlädt, entkoppeln möchten, verbinden Sie sie mit einer asynchronen Warteschlange. In diesem Dokument wird ein Tokio-mpsc-Kanal verwendet. Sie können jedoch jeden Kanal oder jede Warteschlange verwenden, die sich in Ihre asynchrone Laufzeit einbinden lässt.

So passen Sie den Kanalempfänger an eine StreamingSource-Implementierung an:

  1. Definieren Sie eine Struktur, die den Kanal Receiver umschließt: 

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

  2. Implementieren Sie das StreamingSource-Trait für die Wrapper-Struktur, sodass bei jedem Aufruf von next auf den nächsten Block vom Empfänger gewartet wird: 

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

Daten aus der Warteschlange in Cloud Storage streamen

Nachdem Sie das StreamingSource-Trait implementiert haben, können Sie den Objekt-Upload in einer Hintergrundaufgabe starten und Chunks über den Kanal von Ihrer Producer-Aufgabe senden. Da ein Kanalempfänger eine Single-Pass-Datenquelle ist, die Seek nicht implementiert, starten Sie den Upload mit send_buffered anstelle von send_unbuffered.

Führen Sie in einer async-Funktion, die bucket_name im Format projects/_/buckets/BUCKET_ID und object_name akzeptiert, die folgenden Schritte aus:

  1. Initialisieren Sie einen Storage-Client: 

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

  2. Erstellen Sie einen begrenzten mpsc-Kanal, um die Handles sender und receiver abzurufen: 

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

  3. Konfigurieren Sie den Objekthochladevorgang mit QueueSource(receiver) als Nutzlastquelle. Warten Sie noch nicht auf die send_buffered-Zukunft: 

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

  4. Starten Sie eine Hintergrundaufgabe, um den Upload zu steuern, während Ihr Producer Daten generiert: 

    let task = tokio::spawn(upload);

  5. Übertragen Sie in der Producer-Aufgabe Datenblöcke in das sender-Handle: 

    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. Nachdem alle Daten gesendet wurden, löschen Sie das Handle sender, um den Kanal zu schließen und das Ende des Objektstreams zu signalisieren: 

    drop(sender);

  7. Warten Sie auf das Handle für die Hintergrundaufgabe, um zu prüfen, ob sowohl die Aufgabe als auch die Uploadanfrage erfolgreich waren, und rufen Sie die hochgeladenen Object-Metadaten ab: 

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

Bereinigen

Damit Ihrem Google Cloud Konto die in diesem Dokument verwendeten Ressourcen nicht in Rechnung gestellt werden, löschen Sie die Testobjekte, die Sie hochgeladen haben, wenn Sie fertig sind.

Vollständiger Code

Im folgenden Beispiel werden alle vorherigen Schritte zu einer vollständigen Funktion kombiniert:

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