Invia i dati in modalità push sulle scritture degli oggetti Cloud Storage

Il client Cloud Storage nelle librerie client di Cloud per Rust scrive gli oggetti estraendo i blocchi di payload da un tipo che implementa il tratto StreamingSource. Alcune applicazioni generano dati in un'attività o un thread separato e devono invece eseguire il push dei blocchi di payload nel caricamento.

Questo documento mostra come eseguire il wrapping di un ricevitore di canali asincrono in un'implementazione StreamingSource in modo da poter eseguire il push dei dati in una scrittura di oggetti Cloud Storage. Se non hai mai utilizzato la libreria client, consulta prima Utilizzo di Cloud Storage con Rust.

Prima di iniziare

Prima di eseguire il codice in questo documento, completa i seguenti prerequisiti:

• Crea o seleziona un progettoGoogle Cloud con la fatturazione abilitata.
• Crea un bucket Cloud Storage nel tuo progetto.
• Configura le Credenziali predefinite dell'applicazione e concedi alla tua entità un ruolo Identity and Access Management (IAM) con l'autorizzazione per creare oggetti nel bucket, ad esempio Storage Object User (roles/storage.objectUser).

Dipendenze

Aggiungi la libreria client di Cloud Storage e i crate di supporto utilizzati in questo documento al file Cargo.toml:

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

Converti un ricevitore di coda in un'origine di flussi di dati

Per disaccoppiare l'attività che produce i dati da quella che carica il payload, collegale con una coda asincrona. Questo documento utilizza un canale Tokio mpsc, anche se puoi utilizzare qualsiasi canale o coda che si integra con il tuo runtime asincrono.

Per adattare il ricevitore del canale a un'implementazione StreamingSource, completa i seguenti passaggi:

1. Definisci una struttura che aggrega il canale Receiver:

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

2. Implementa il tratto StreamingSource per la struttura wrapper in modo che ogni chiamata a next attenda il chunk successivo dal ricevitore:

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

Trasmetti i dati dalla coda a Cloud Storage

Dopo aver implementato il tratto StreamingSource, puoi generare il caricamento dell'oggetto in un'attività in background e inviare i blocchi tramite il canale dall'attività del producer. Poiché un ricevitore di canale è un'origine dati a passaggio singolo che non implementa Seek, avvia il caricamento utilizzando send_buffered anziché send_unbuffered.

All'interno di una funzione async che accetta bucket_name nel formato projects/_/buckets/BUCKET_ID e object_name, completa i seguenti passaggi:

1. Inizializza un client Storage:

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

2. Crea un canale mpsc limitato per ottenere gli handle sender e receiver:

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

3. Configura il caricamento dell'oggetto con QueueSource(receiver) come origine del payload. Non aspettare ancora il futuro di send_buffered:

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

4. Genera un'attività in background per gestire il caricamento mentre il producer genera i dati:

   let task = tokio::spawn(upload);

5. Nell'attività producer, esegui il push dei blocchi di dati nell'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. Dopo aver inviato tutti i dati, elimina l'handle sender per chiudere il canale e segnalare la fine dello stream di oggetti:

   drop(sender);

7. Attendi l'handle dell'attività in background per verificare che l'attività e la richiesta di caricamento siano andate a buon fine e recupera i metadati Object caricati:

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

Nota: se il caricamento non va a buon fine prima dell'invio di tutti i blocchi, l'attività in background viene chiusa e il ricevitore viene eliminato, causando la restituzione di un errore di canale chiuso da parte di sender.send. Nel codice di produzione, se sender.send non va a buon fine, attendi l'handle dell'attività in background per ispezionare l'errore Cloud Storage sottostante.

Esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo documento, elimina gli oggetti di test che hai caricato al termine.

Completamento codice

L'esempio seguente combina tutti i passaggi precedenti in una funzione 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(())
}