Envoyer des données lors des écritures d'objets Cloud Storage

Le client Cloud Storage des bibliothèques clientes Cloud pour Rust écrit des objets en extrayant des fragments de charge utile d'un type qui implémente la caractéristique StreamingSource. Certaines applications génèrent des données dans une tâche ou un thread distinct, et doivent plutôt transférer des fragments de charge utile vers l'importation.

Ce document explique comment encapsuler un récepteur de canal asynchrone dans une implémentation StreamingSource afin de pouvoir transférer des données vers une écriture d'objet Cloud Storage. Si vous débutez avec la bibliothèque cliente, consultez d'abord Utiliser Cloud Storage avec Rust.

Avant de commencer

Avant d'exécuter le code de ce document, assurez-vous de remplir les conditions préalables suivantes :

Dépendances

Ajoutez la bibliothèque cliente Cloud Storage et les crates compatibles utilisées dans ce document à votre fichier Cargo.toml :

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

Convertir un récepteur de file d'attente en source de streaming

Pour dissocier la tâche qui produit les données de celle qui importe la charge utile, connectez-les à l'aide d'une file d'attente asynchrone. Ce document utilise un canal Tokio mpsc, mais vous pouvez utiliser n'importe quel canal ou file d'attente qui s'intègre à votre environnement d'exécution asynchrone.

Pour adapter le récepteur de canal à une implémentation StreamingSource, procédez comme suit :

  1. Définissez une structure qui encapsule le canal Receiver : 

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

  2. Implémentez la caractéristique StreamingSource pour la structure du wrapper afin que chaque appel à next attende le prochain fragment du récepteur : 

    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)
    }
}

Transférer des données de la file d'attente vers Cloud Storage

Après avoir implémenté la caractéristique StreamingSource, vous pouvez générer l'importation d'objets dans une tâche en arrière-plan et envoyer des fragments via le canal à partir de votre tâche de producteur. Étant donné qu'un récepteur de canal est une source de données à un seul passage qui n'implémente pas Seek, lancez l'importation à l'aide de send_buffered plutôt que de send_unbuffered.

Dans une fonction async qui accepte bucket_name au format projects/_/buckets/BUCKET_ID et object_name, procédez comme suit :

  1. Initialisez un client Storage : 

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

  2. Créez un canal mpsc limité pour obtenir les poignées sender et receiver : 

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

  3. Configurez l'importation d'objets avec QueueSource(receiver) comme source de charge utile. N'attendez pas encore le futur send_buffered : 

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

  4. Lancez une tâche en arrière-plan pour gérer l'importation pendant que votre producteur génère des données : 

    let task = tokio::spawn(upload);

  5. Dans la tâche du producteur, transmettez des fragments de données au handle 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. Après avoir envoyé toutes les données, supprimez le handle sender pour fermer le canal et signaler la fin du flux d'objets : 

    drop(sender);

  7. Attendez le handle de la tâche en arrière-plan pour vérifier que la tâche et la requête d'importation ont réussi, puis récupérez les métadonnées Object importées : 

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

Effectuer un nettoyage

Pour éviter que les ressources utilisées dans ce document ne soient facturées sur votre compte Google Cloud , supprimez les objets de test que vous avez importés lorsque vous avez terminé.

Compléter le code

L'exemple suivant combine toutes les étapes précédentes dans une fonction complète :

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(())
}