ביטול כתיבה של אובייקטים ב-Cloud Storage עם שגיאות

הלקוח של Cloud Storage בספריות הלקוח ב-Cloud ל-Rust כותב אובייקטים על ידי משיכת קטעי מטען ייעודי (payload) מסוג שמטמיע את המאפיין StreamingSource. אם האפליקציה מזהה מצב שגיאה בזמן יצירה או סטרימינג של נתונים, החזרת שגיאה מההטמעה של StreamingSource מונעת מספריית הלקוח להשלים את יצירת האובייקט.

במסמך הזה מוסבר איך להגדיר סוג שגיאה בהתאמה אישית ולהטמיע את StreamingSource כדי להחזיר את השגיאה הזו באמצע הסטרימינג. אתם גם מוודאים שההעלאה שנכשלה מחזירה את השגיאה שלכם בלי ליצור את האובייקט. אם אתם משתמשים חדשים בספריית הלקוח, קודם כדאי לקרוא את המאמר שימוש ב-Cloud Storage עם Rust.

לפני שמתחילים

לפני שמריצים את הקוד במסמך הזה, צריך להשלים את הדרישות המוקדמות הבאות:

יחסי תלות

מוסיפים את ספריית הלקוח של Cloud Storage ואת ה-crates התומכים לקובץ Cargo.toml:

cargo add google-cloud-storage anyhow bytes

יצירת סוג שגיאה בהתאמה אישית

כדי לבטל כתיבה של אובייקט לפני הסיום, ההטמעה של StreamingSource צריכה להחזיר שגיאה שמטמיעה את std::error::Error + Send + Sync + 'static. אפשר להשתמש בכל סוג שגיאה קיים שעומד בגבולות האלה, או להגדיר enum בהתאמה אישית לאפליקציה. כדי להגדיר טיפוס enum מותאם אישית של שגיאות:

  1. מגדירים enum של שגיאה בהתאמה אישית (MyError): 

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

  2. מטמיעים את המאפיין הרגיל Error כדי שספריית הלקוח תוכל לשמור את השגיאה כסיבה הבסיסית: 

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

  3. מטמיעים את המאפיין Display שנדרש על ידי 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"),
        }
    }
}

הטמעה של מקור סטרימינג בהתאמה אישית

אחר כך יוצרים סוג שמייצר קטעים סינתטיים ומחזיר שגיאה אחרי 41 קטעים, כדי שההעלאה תיפסק לפני שהיא תסתיים. כך עושים את זה:

  1. מגדירים מבנה למעקב אחרי מונה החלקים: 

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

  2. מטמיעים את StreamingSource עבור MySource, משייכים את type Error = MyError ומחזירים את Some(Err(MyError::ExpectedProblem)) באיטרציה 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"
            )))),
        }
    }
}

כש-next מחזיר את הערך Some(Err(...)), ספריית הלקוח מפסיקה לקרוא מהמקור ומבטלת את בקשת הכתיבה בלי להשלים את האובייקט.

מנסים להעלות את הקובץ ובודקים את השגיאה

אחרי שמטמיעים את StreamingSource, מנסים לכתוב אובייקט במאגר נתונים זמני ומוודאים שהלקוח מחזיר שגיאת סריאליזציה שעוטפת את MyError. מכיוון ש-MySource הוא מקור במעבר יחיד שלא מטמיע את Seek, צריך להתחיל את הכתיבה באמצעות send_buffered במקום send_unbuffered.

בתוך פונקציית async שמקבלת את bucket_name בפורמט projects/_/buckets/BUCKET_ID, מבצעים את הפעולות הבאות:

  1. מאתחלים לקוח Storage: 

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

  2. מתחילים את הכתיבה של האובייקט במאגר הזמני באמצעות MySource::default() כמטען הייעודי (payload) של המקור: 

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

  3. מוודאים שההעלאה נכשלת עם שגיאת סריאליזציה שמקור השגיאה שלה הוא 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. מחזירים את Ok(()) כדי להשלים את הפונקציה: 

        Ok(())
}

כשמריצים את attempt_upload, מודפסת הודעת Upload result Err(...) שמכילה את MyError::ExpectedProblem. הטענות הנכונות מאמתות שהלקוח הפסיק את ההעלאה והחזיר שגיאת סריאליזציה שעוטפת את MyError. מכיוון שהכתיבה בוטלה לפני הסיום, Cloud Storage לא יוצר את האובייקט expect-error בקטגוריה.

הסרת המשאבים

למרות שהכתיבה שבוטלה לא יוצרת אובייקט, מחקו את כל אובייקטי הבדיקה שיצרתם בשביל המסמך הזה. כדי להימנע מחיובים מתמשכים, מומלץ למחוק את משאבי הבדיקה בסיום.

השלמת הקוד

בדוגמה הבאה משולבים כל השלבים הקודמים במודול שלם:

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

המאמרים הבאים