Cancelar gravações de objetos do Cloud Storage com erros

O cliente do Cloud Storage nas bibliotecas de cliente do Cloud para Rust grava objetos extraindo blocos de payload de um tipo que implementa a característica StreamingSource. Se o aplicativo detectar uma condição de erro ao gerar ou transmitir dados, o retorno de um erro da implementação de StreamingSource impedirá que a biblioteca de cliente finalize a criação do objeto.

Neste documento, mostramos como definir um tipo de erro personalizado e implementar StreamingSource para retornar esse erro no meio do stream. Você também verifica se o upload com falha retorna o erro sem criar o objeto. Se você não conhece a biblioteca de cliente, consulte Como usar o Cloud Storage com Rust primeiro.

Antes de começar

Antes de executar o código neste documento, conclua os seguintes pré-requisitos:

Dependências

Adicione a biblioteca de cliente do Cloud Storage e os crates de suporte ao arquivo Cargo.toml:

cargo add google-cloud-storage anyhow bytes

Criar um tipo de erro personalizado

Para cancelar uma gravação de objeto antes da finalização, sua implementação de StreamingSource precisa retornar um erro que implemente std::error::Error + Send + Sync + 'static. É possível usar qualquer tipo de erro que atenda a esses limites ou definir um tipo enumerado personalizado para o aplicativo. Siga as etapas abaixo para definir um tipo enumerado de erro personalizado:

  1. Defina um tipo enumerado de erro personalizado (MyError): 

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

  2. Implemente a característica padrão Error para que a biblioteca de cliente possa preservar o erro como a causa subjacente: 

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

  3. Implemente a característica Display exigida por 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"),
        }
    }
}

Implementar uma fonte de streaming personalizada

Em seguida, crie um tipo que gere blocos sintéticos e retorne um erro após 41 blocos para que o upload seja interrompido antes da conclusão. Siga estas etapas:

  1. Defina uma struct para rastrear o contador de blocos: 

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

  2. Implemente StreamingSource para MySource, associando type Error = MyError e retornando Some(Err(MyError::ExpectedProblem)) na iteração 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 retorna Some(Err(...)), a biblioteca de cliente para de ler da origem e cancela a solicitação de gravação sem finalizar o objeto.

Tentar fazer o upload e inspecionar o erro

Depois de implementar StreamingSource, tente uma gravação de objeto em buffer e verifique se o cliente retorna um erro de serialização que envolve MyError. Como MySource é uma fonte de passagem única que não implementa Seek, inicie a gravação com send_buffered em vez de send_unbuffered.

Em uma função async que aceita bucket_name no formato projects/_/buckets/BUCKET_ID, siga estas etapas:

  1. Inicialize um cliente Storage: 

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

  2. Inicie a gravação do objeto em buffer usando MySource::default() como a origem do payload: 

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

  3. Verifique se o upload falha com um erro de serialização em que o erro de origem é 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. Retorne Ok(()) para concluir a função: 

        Ok(())
}

Quando você executa attempt_upload, ele mostra uma mensagem Upload result Err(...) contendo MyError::ExpectedProblem. As declarações verificam se o cliente interrompeu o upload e retornou um erro de serialização que envolve MyError. Como a gravação foi cancelada antes da finalização, o Cloud Storage não cria o objeto expect-error no bucket.

Limpar

Embora a gravação cancelada não crie um objeto, exclua todos os objetos de teste criados para este documento. Excluir os recursos de teste quando terminar evita cobranças contínuas.

Preencher código

O exemplo a seguir combina todas as etapas anteriores em um módulo 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(())
}

A seguir