Annuler les écritures d'objets Cloud Storage comportant des erreurs

Le client Cloud Storage des bibliothèques clientes Cloud pour Rust écrit des objets en extrayant des fragments de charge utile d'un type qui implémente la caractéristique StreamingSource. Si votre application détecte une condition d'erreur lors de la génération ou de la diffusion de données, le renvoi d'une erreur à partir de votre implémentation StreamingSource empêche la bibliothèque cliente de finaliser la création de l'objet.

Ce document explique comment définir un type d'erreur personnalisé et implémenter StreamingSource pour renvoyer cette erreur en cours de flux. Vous vérifiez également que l'échec de l'importation renvoie votre erreur sans créer l'objet. Si vous débutez avec la bibliothèque cliente, consultez d'abord Utiliser Cloud Storage avec Rust.

Avant de commencer

Avant d'exécuter le code de ce document, assurez-vous de remplir les conditions préalables suivantes :

Dépendances

Ajoutez la bibliothèque cliente Cloud Storage et les crates compatibles à votre fichier Cargo.toml :

cargo add google-cloud-storage anyhow bytes

Créer un type d'erreur personnalisé

Pour annuler l'écriture d'un objet avant sa finalisation, votre implémentation StreamingSource doit renvoyer une erreur qui implémente std::error::Error + Send + Sync + 'static. Vous pouvez utiliser n'importe quel type d'erreur existant qui respecte ces limites ou définir une énumération personnalisée pour votre application. Pour définir une énumération d'erreur personnalisée, procédez comme suit :

  1. Définissez une énumération d'erreur personnalisée (MyError) : 

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

  2. Implémentez la caractéristique Error standard afin que la bibliothèque cliente puisse conserver votre erreur en tant que cause sous-jacente : 

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

  3. Implémentez la caractéristique Display requise par 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"),
        }
    }
}

Implémenter une source de streaming personnalisée

Ensuite, créez un type qui génère des fragments synthétiques et renvoie une erreur après 41 fragments, de sorte que l'importation s'arrête avant la fin. Procédez comme suit :

  1. Définissez une structure pour suivre le compteur de fragments : 

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

  2. Implémentez StreamingSource pour MySource, en associant type Error = MyError et en renvoyant Some(Err(MyError::ExpectedProblem)) à l'itération 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"
            )))),
        }
    }
}

Lorsque next renvoie Some(Err(...)), la bibliothèque cliente arrête de lire la source et annule la requête d'écriture sans finaliser l'objet.

Tenter d'importer et inspecter l'erreur

Après avoir implémenté StreamingSource, tentez d'écrire un objet mis en mémoire tampon et vérifiez que le client renvoie une erreur de sérialisation encapsulant MyError. Étant donné que MySource est une source à un seul passage qui n'implémente pas Seek, démarrez l'écriture avec send_buffered au lieu de send_unbuffered.

Dans une fonction async qui accepte bucket_name au format projects/_/buckets/BUCKET_ID, procédez comme suit :

  1. Initialisez 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. Démarrez l'écriture d'objet mis en mémoire tampon en utilisant MySource::default() comme source de charge utile : 

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

  3. Vérifiez que l'importation échoue avec une erreur de sérialisation dont l'erreur source est 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. Renvoyez Ok(()) pour terminer la fonction : 

        Ok(())
}

Lorsque vous exécutez attempt_upload, un message Upload result Err(...) contenant MyError::ExpectedProblem s'affiche. Les assertions vérifient que le client a arrêté l'importation et a renvoyé une erreur de sérialisation encapsulant MyError. Comme l'écriture a été annulée avant la finalisation, Cloud Storage ne crée pas l'objet expect-error dans le bucket.

Effectuer un nettoyage

Même si l'écriture annulée ne crée pas d'objet, supprimez tous les objets de test que vous avez créés pour ce document. La suppression des ressources de test une fois que vous avez terminé vous évite des frais récurrents.

Compléter le code

L'exemple suivant combine toutes les étapes précédentes dans un module complet :

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

Étapes suivantes