Schreibvorgänge für Cloud Storage-Objekte mit Fehlern abbrechen

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. Wenn Ihre Anwendung beim Generieren oder Streamen von Daten einen Fehler erkennt, wird die Clientbibliothek durch die Rückgabe eines Fehlers aus Ihrer StreamingSource-Implementierung daran gehindert, die Objekterstellung abzuschließen.

In diesem Dokument wird beschrieben, wie Sie einen benutzerdefinierten Fehlertyp definieren und StreamingSource implementieren, um diesen Fehler während des Streams zurückzugeben. Außerdem prüfen Sie, ob der fehlgeschlagene Upload Ihren Fehler zurückgibt, ohne das Objekt zu erstellen. 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 unterstützenden Crates der Datei Cargo.toml hinzu:

cargo add google-cloud-storage anyhow bytes

Benutzerdefinierten Fehlertyp erstellen

Wenn Sie einen Objektschreibvorgang vor dem Abschluss abbrechen möchten, muss Ihre StreamingSource-Implementierung einen Fehler zurückgeben, der std::error::Error + Send + Sync + 'static implementiert. Sie können einen beliebigen vorhandenen Fehlertyp verwenden, der diese Grenzen erfüllt, oder ein benutzerdefiniertes Enum für Ihre Anwendung definieren. Führen Sie die folgenden Schritte aus, um ein benutzerdefiniertes Fehler-Enum zu definieren:

  1. Definieren Sie ein benutzerdefiniertes Fehler-Enum (MyError): 

    #[derive(Debug)]
pub enum MyError {
    ExpectedProblem,
    OhNoes,
}

  2. Implementieren Sie das Standardmerkmal Error, damit die Clientbibliothek den Fehler als zugrunde liegende Ursache beibehalten kann: 

    impl std::error::Error for MyError {}

  3. Implementiere das für std::error::Error erforderliche Display-Attribut: 

    impl std::fmt::Display for MyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ExpectedProblem => write!(f, "this kind of thing happens"),
            Self::OhNoes => write!(f, "oh noes! something terrible happened"),
        }
    }
}

Benutzerdefinierte Streamingquelle implementieren

Erstellen Sie als Nächstes einen Typ, der synthetische Blöcke generiert und nach 41 Blöcken einen Fehler zurückgibt, damit der Upload vor dem Abschluss beendet wird. Gehen Sie folgendermaßen vor:

  1. Definieren Sie eine Struktur, um den Chunk-Zähler zu verfolgen: 

    #[derive(Debug, Default)]
struct MySource(u32);

  2. Implementieren Sie StreamingSource für MySource, verknüpfen Sie type Error = MyError und geben Sie Some(Err(MyError::ExpectedProblem)) bei Iteration 42 zurück: 

    impl google_cloud_storage::streaming_source::StreamingSource for MySource {
    type Error = MyError;
    async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
        self.0 += 1;
        match self.0 {
            42 => Some(Err(MyError::ExpectedProblem)),
            n if n > 42 => None,
            n => Some(Ok(bytes::Bytes::from_owner(format!(
                "test data for the example {n}\n"
            )))),
        }
    }
}

Wenn next den Wert Some(Err(...)) zurückgibt, liest die Clientbibliothek nicht mehr aus der Quelle und bricht die Schreibanfrage ab, ohne das Objekt abzuschließen.

Upload versuchen und Fehler prüfen

Versuchen Sie nach der Implementierung von StreamingSource, ein gepuffertes Objekt zu schreiben, und prüfen Sie, ob der Client einen Serialisierungsfehler zurückgibt, der MyError umschließt. Da MySource eine Single-Pass-Quelle ist, die Seek nicht implementiert, starten Sie den Schreibvorgang mit send_buffered anstelle von send_unbuffered.

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

  1. Initialisieren Sie einen Storage-Client: 

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

  2. Starten Sie den gepufferten Objektschreibvorgang mit MySource::default() als Nutzlastquelle: 

    let upload = client
    .write_object(bucket_name, "expect-error", MySource::default())
    .send_buffered()
    .await;

  3. Prüfen Sie, ob der Upload mit einem Serialisierungsfehler fehlschlägt, dessen Quellfehler MyError ist: 

    println!("Upload result {upload:?}");
let err = upload.expect_err("the source is supposed to terminate the upload");
assert!(err.is_serialization(), "{err:?}");
use std::error::Error as _;
assert!(err.source().is_some_and(|e| e.is::<MyError>()), "{err:?}");

  4. Drücken Sie Ok(()), um die Funktion abzuschließen: 

        Ok(())
}

Wenn Sie attempt_upload ausführen, wird eine Upload result Err(...)-Nachricht mit MyError::ExpectedProblem ausgegeben. Die Assertions bestätigen, dass der Client den Upload beendet und einen Serialisierungsfehler zurückgegeben hat, der MyError umschließt. Da der Schreibvorgang vor dem Abschluss abgebrochen wurde, erstellt Cloud Storage das Objekt expect-error nicht im Bucket.

Bereinigen

Auch wenn durch den abgebrochenen Schreibvorgang kein Objekt erstellt wird, sollten Sie alle Testobjekte löschen, die Sie für dieses Dokument erstellt haben. Wenn Sie die Testressourcen nach Abschluss löschen, vermeiden Sie laufende Kosten.

Vollständiger Code

Im folgenden Beispiel werden alle vorherigen Schritte zu einem vollständigen Modul kombiniert:

#[derive(Debug)]
pub enum MyError {
    ExpectedProblem,
    OhNoes,
}

impl std::error::Error for MyError {}

impl std::fmt::Display for MyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ExpectedProblem => write!(f, "this kind of thing happens"),
            Self::OhNoes => write!(f, "oh noes! something terrible happened"),
        }
    }
}

#[derive(Debug, Default)]
struct MySource(u32);

impl google_cloud_storage::streaming_source::StreamingSource for MySource {
    type Error = MyError;
    async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
        self.0 += 1;
        match self.0 {
            42 => Some(Err(MyError::ExpectedProblem)),
            n if n > 42 => None,
            n => Some(Ok(bytes::Bytes::from_owner(format!(
                "test data for the example {n}\n"
            )))),
        }
    }
}

pub async fn attempt_upload(bucket_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::client::Storage;
    let client = Storage::builder().build().await?;
    let upload = client
        .write_object(bucket_name, "expect-error", MySource::default())
        .send_buffered()
        .await;
    println!("Upload result {upload:?}");
    let err = upload.expect_err("the source is supposed to terminate the upload");
    assert!(err.is_serialization(), "{err:?}");
    use std::error::Error as _;
    assert!(err.source().is_some_and(|e| e.is::<MyError>()), "{err:?}");
    Ok(())
}

Nächste Schritte