取消 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 trait： 

    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 不会在存储桶中创建 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(())
}

后续步骤