Annulla le scritture di oggetti Cloud Storage con errori

Il client Cloud Storage nelle librerie client di Cloud per Rust scrive gli oggetti estraendo i blocchi di payload da un tipo che implementa il tratto StreamingSource. Se l'applicazione rileva una condizione di errore durante la generazione o lo streaming dei dati, la restituzione di un errore dall'implementazione di StreamingSource impedisce alla libreria client di finalizzare la creazione dell'oggetto.

Questo documento mostra come definire un tipo di errore personalizzato e implementare StreamingSource per restituire l'errore durante lo stream. Verifichi inoltre che il caricamento non riuscito restituisca l'errore senza creare l'oggetto. Se non hai mai utilizzato la libreria client, consulta prima Utilizzo di Cloud Storage con Rust.

Prima di iniziare

Prima di eseguire il codice in questo documento, completa i seguenti prerequisiti:

Dipendenze

Aggiungi la libreria client di Cloud Storage e i crate di supporto al file Cargo.toml:

cargo add google-cloud-storage anyhow bytes

Crea un tipo di errore personalizzato

Per annullare la scrittura di un oggetto prima della finalizzazione, l'implementazione di StreamingSource deve restituire un errore che implementa std::error::Error + Send + Sync + 'static. Puoi utilizzare qualsiasi tipo di errore esistente che soddisfi questi limiti o definire un enum personalizzato per la tua applicazione. Per definire un enum di errore personalizzato, segui questi passaggi:

  1. Definisci un enum di errore personalizzato (MyError): 

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

  2. Implementa il tratto standard Error in modo che la libreria client possa conservare l'errore come causa sottostante: 

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

  3. Implementa il tratto Display richiesto da std::error::Error: 

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

Implementa un'origine di streaming personalizzata

Successivamente, crea un tipo che genera chunk sintetici e restituisce un errore dopo 41 chunk, in modo che il caricamento si interrompa prima del completamento. Completa i seguenti passaggi:

  1. Definisci una struttura per monitorare il contatore dei blocchi: 

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

  2. Implementa StreamingSource per MySource, associando type Error = MyError e restituendo Some(Err(MyError::ExpectedProblem)) all'iterazione 42: 

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

Quando next restituisce Some(Err(...)), la libreria client interrompe la lettura dall'origine e annulla la richiesta di scrittura senza finalizzare l'oggetto.

Prova a eseguire il caricamento e controlla l'errore

Dopo aver implementato StreamingSource, prova a eseguire una scrittura di oggetti con buffer e verifica che il client restituisca un errore di serializzazione che racchiude MyError. Poiché MySource è un'origine a passaggio singolo che non implementa Seek, avvia la scrittura con send_buffered anziché con send_unbuffered.

All'interno di una funzione async che accetta bucket_name nel formato projects/_/buckets/BUCKET_ID, completa i seguenti passaggi:

  1. Inizializza un client Storage: 

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

  2. Avvia la scrittura dell'oggetto con buffer utilizzando MySource::default() come origine del payload: 

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

  3. Verifica che il caricamento non vada a buon fine con un errore di serializzazione il cui errore di origine è MyError: 

    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. Restituisci Ok(()) per completare la funzione: 

        Ok(())
}

Quando esegui attempt_upload, viene stampato un messaggio Upload result Err(...) contenente MyError::ExpectedProblem. Le asserzioni verificano che il client abbia interrotto il caricamento e restituito un errore di serializzazione che include MyError. Poiché la scrittura è stata annullata prima della finalizzazione, Cloud Storage non crea l'oggetto expect-error nel bucket.

Esegui la pulizia

Anche se la scrittura annullata non crea un oggetto, elimina tutti gli oggetti di test che hai creato per questo documento. L'eliminazione delle risorse di test al termine evita addebiti continui.

Completamento codice

L'esempio seguente combina tutti i passaggi precedenti in un modulo completo:

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

Passaggi successivi