Enviar dados em gravações de objetos do Cloud Storage

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. Alguns aplicativos geram dados em uma tarefa ou linha de execução separada e precisam enviar blocos de payload para o upload.

Neste documento, mostramos como encapsular um receptor de canal assíncrono em uma implementação de StreamingSource para que seja possível enviar dados para uma gravação de objeto do Cloud Storage. 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:

• Crie ou selecione um projeto doGoogle Cloud com o faturamento ativado.
• Crie um bucket do Cloud Storage no projeto.
• Configure o Application Default Credentials e conceda ao principal um papel do Identity and Access Management (IAM) com permissão para criar objetos no bucket, como Usuário de objetos do Storage (roles/storage.objectUser).

Dependências

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

cargo add google-cloud-storage anyhow bytes
cargo add tokio --features rt,sync

Converter um receptor de fila em uma fonte de streaming

Para separar a tarefa que produz dados da tarefa que faz upload do payload, conecte-as com uma fila assíncrona. Neste documento, usamos um canal mpsc do Tokio, mas é possível usar qualquer canal ou fila que se integre ao ambiente de execução assíncrono.

Para adaptar o receptor de canal em uma implementação de StreamingSource, siga estas etapas:

1. Defina uma struct que encapsule o canal Receiver:

   use tokio::sync::mpsc::{self, Receiver};
   #[derive(Debug)]
   struct QueueSource(Receiver<bytes::Bytes>);

2. Implemente a característica StreamingSource para a struct do wrapper para que cada chamada para next aguarde o próximo bloco do receptor:

   use google_cloud_storage::streaming_source::StreamingSource;
   impl StreamingSource for QueueSource {
       type Error = std::convert::Infallible;
       async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
           self.0.recv().await.map(Ok)
       }
   }

Transmitir dados da fila para o Cloud Storage

Depois de implementar a característica StreamingSource, é possível gerar o upload do objeto em uma tarefa em segundo plano e enviar blocos pelo canal da tarefa do produtor. Como um receptor de canal é uma fonte de dados de passagem única que não implementa Seek, inicie o upload usando send_buffered em vez de send_unbuffered.

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

1. Inicialize um cliente Storage:

   use google_cloud_storage::client::Storage;
   let client = Storage::builder().build().await?;

2. Crie um canal mpsc limitado para receber os identificadores sender e receiver:

   let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);

3. Configure o upload do objeto com QueueSource(receiver) como a origem do payload. Ainda não aguarde o futuro de send_buffered:

   let upload = client
       .write_object(bucket_name, object_name, QueueSource(receiver))
       .send_buffered();

4. Gere uma tarefa em segundo plano para impulsionar o upload enquanto o produtor gera dados:

   let task = tokio::spawn(upload);

5. Na tarefa do produtor, envie blocos de dados para o identificador sender:

   for _ in 0..1000 {
       let line = "I will not write funny examples in class\n";
       sender
           .send(bytes::Bytes::from_static(line.as_bytes()))
           .await?;
   }

6. Depois de enviar todos os dados, solte o identificador sender para fechar o canal e sinalizar o fim do fluxo de objetos:

   drop(sender);

7. Aguarde o identificador da tarefa em segundo plano para verificar se a tarefa e a solicitação de upload foram bem-sucedidas e recupere os metadados de Object enviados:

   let object = task.await??;
   println!("object successfully uploaded {object:?}");

Observação: se o upload falhar antes que todos os blocos sejam enviados, a tarefa em segundo plano será encerrada e descartará o receptor, fazendo com que sender.send retorne um erro de canal fechado. No código de produção, se sender.send falhar, aguarde o identificador da tarefa em segundo plano para inspecionar o erro do Cloud Storage.

Limpar

Para evitar cobranças na sua conta do Google Cloud pelos recursos usados neste documento, exclua os objetos de teste que você enviou quando terminar.

Preencher código

O exemplo a seguir combina todas as etapas anteriores em uma função completa:

use google_cloud_storage::streaming_source::StreamingSource;
use tokio::sync::mpsc::{self, Receiver};
#[derive(Debug)]
struct QueueSource(Receiver<bytes::Bytes>);
impl StreamingSource for QueueSource {
    type Error = std::convert::Infallible;
    async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
        self.0.recv().await.map(Ok)
    }
}

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

    let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);
    let upload = client
        .write_object(bucket_name, object_name, QueueSource(receiver))
        .send_buffered();
    let task = tokio::spawn(upload);

    for _ in 0..1000 {
        let line = "I will not write funny examples in class\n";
        sender
            .send(bytes::Bytes::from_static(line.as_bytes()))
            .await?;
    }
    drop(sender);
    let object = task.await??;
    println!("object successfully uploaded {object:?}");

    Ok(())
}