Envía datos sobre las escrituras de objetos de Cloud Storage

El cliente de Cloud Storage en las bibliotecas cliente de Cloud para Rust escribe objetos extrayendo fragmentos de carga útil de un tipo que implementa el atributo StreamingSource. Algunas aplicaciones generan datos en una tarea o un subproceso independientes y, en su lugar, necesitan enviar fragmentos de carga útil a la carga.

En este documento, se muestra cómo unir un receptor de canal asíncrono en una implementación de StreamingSource para que puedas enviar datos a una escritura de objetos de Cloud Storage. Si es la primera vez que usas la biblioteca cliente, consulta Usa Cloud Storage con Rust primero.

Antes de comenzar

Antes de ejecutar el código de este documento, completa los siguientes requisitos previos:

• Crea o selecciona un proyecto deGoogle Cloud con la facturación habilitada.
• Crea un bucket de Cloud Storage en tu proyecto.
• Configura las credenciales predeterminadas de la aplicación y otorga a tu principal un rol de Identity and Access Management (IAM) con permiso para crear objetos en el bucket, como Usuario de objetos de almacenamiento (roles/storage.objectUser).

Dependencias

Agrega la biblioteca cliente de Cloud Storage y los crates de apoyo que se usan en este documento a tu archivo Cargo.toml:

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

Convierte un receptor de cola en una fuente de transmisión

Para desvincular la tarea que produce datos de la tarea que sube la carga útil, conéctalas con una cola asíncrona. En este documento, se usa un canal mpsc de Tokio, aunque puedes usar cualquier canal o cola que se integre en tu entorno de ejecución asíncrono.

Para adaptar el receptor de canales a una implementación de StreamingSource, completa los siguientes pasos:

1. Define un struct que una el canal Receiver:

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

2. Implementa el atributo StreamingSource para la estructura de wrapper de modo que cada llamada a next espere el siguiente fragmento del 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)
       }
   }

Transmite datos de la cola a Cloud Storage

Después de implementar el atributo StreamingSource, puedes generar la carga del objeto en una tarea en segundo plano y enviar fragmentos a través del canal desde tu tarea de productor. Debido a que un receptor de canal es una fuente de datos de un solo paso que no implementa Seek, inicia la carga con send_buffered en lugar de send_unbuffered.

Dentro de una función async que acepte bucket_name en el formato projects/_/buckets/BUCKET_ID y object_name, completa los siguientes pasos:

1. Inicializa un cliente de Storage:

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

2. Crea un canal mpsc limitado para obtener los controladores sender y receiver:

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

3. Configura la carga de objetos con QueueSource(receiver) como la fuente de la carga útil. Aún no esperes el futuro de send_buffered:

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

4. Genera una tarea en segundo plano para controlar la carga mientras tu productor genera datos:

   let task = tokio::spawn(upload);

5. En la tarea del productor, envía fragmentos de datos al controlador 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. Después de enviar todos los datos, descarta el controlador sender para cerrar el canal y marcar el final del flujo de objetos:

   drop(sender);

7. Espera el controlador de tareas en segundo plano para verificar que la tarea y la solicitud de carga se hayan realizado correctamente, y recupera los metadatos de Object subidos:

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

Nota: Si la carga falla antes de que se envíen todos los fragmentos, la tarea en segundo plano se cierra y se descarta el receptor, lo que hace que sender.send devuelva un error de canal cerrado. En el código de producción, si sender.send falla, espera el controlador de tareas en segundo plano para inspeccionar el error subyacente de Cloud Storage.

Realiza una limpieza

Para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos que usaste en este documento, borra los objetos de prueba que subiste cuando termines.

Completar código

En el siguiente ejemplo, se combinan todos los pasos anteriores en una función 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(())
}