取消寫入 Cloud Storage 物件，並顯示錯誤

Rust 適用的 Cloud 用戶端程式庫中的 Cloud Storage 用戶端，會從實作 StreamingSource 特徵的類型提取酬載區塊，藉此寫入物件。如果應用程式在產生或串流資料時偵測到錯誤狀況，從 StreamingSource 實作傳回錯誤會阻止用戶端程式庫完成物件建立。

本文說明如何定義自訂錯誤類型，並實作 StreamingSource，在串流中傳回該錯誤。您也確認上傳失敗時會傳回錯誤，不會建立物件。如果您是第一次使用用戶端程式庫，請先參閱「搭配使用 Cloud Storage 和 Rust」。

事前準備

執行本文件中的程式碼前，請先完成下列必要條件：

依附元件

將 Cloud Storage 用戶端程式庫和支援的 Crate 新增至 Cargo.toml 檔案：

cargo add google-cloud-storage anyhow bytes

建立自訂錯誤類型

如要在完成前取消物件寫入，StreamingSource 實作必須傳回實作 std::error::Error + Send + Sync + 'static 的錯誤。您可以使用符合這些界限的任何現有錯誤類型，或為應用程式定義自訂列舉。請按照下列步驟定義自訂錯誤列舉：

  1. 定義自訂錯誤列舉 (MyError)： 

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

  2. 實作標準 Error 特徵，讓用戶端程式庫將錯誤保留為根本原因： 

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

  3. 實作 std::error::Error 所需的 Display 特徵：

    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. 為 MySource 實作 StreamingSource，在第 42 次疊代時建立 type Error = MyError 關聯並傳回 Some(Err(MyError::ExpectedProblem))： 

    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 開始寫入。

在接受 projects/_/buckets/BUCKET_ID 格式 bucket_name 的 async 函式中，完成下列步驟：

  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() 做為酬載來源，開始寫入緩衝物件： 

    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 時，系統會列印包含 MyError::ExpectedProblem 的 Upload result Err(...) 訊息。這些斷言會驗證用戶端是否停止上傳，並傳回包含 MyError 的序列化錯誤。由於寫入作業在完成前取消，Cloud Storage 不會在 bucket 中建立 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(())
}

後續步驟