Google은 AI 기술을 사용하여 콘텐츠를 사용자의 기본 언어로 번역합니다. AI 번역에는 오류가 있을 수 있습니다.

Cloud Storage 객체 쓰기에 대한 데이터 푸시

Rust용 Cloud 클라이언트 라이브러리의 Cloud Storage 클라이언트는 StreamingSource 특성을 구현하는 유형에서 페이로드 청크를 가져와 객체를 작성합니다. 일부 애플리케이션은 별도의 태스크 또는 스레드에서 데이터를 생성하며 대신 페이로드 청크를 업로드에 푸시해야 합니다.

이 문서에서는 Cloud Storage 객체 쓰기에 데이터를 푸시할 수 있도록 StreamingSource 구현에서 비동기 채널 수신자를 래핑하는 방법을 보여줍니다. 클라이언트 라이브러리를 처음 사용하는 경우 먼저 Rust에서 Cloud Storage 사용을 참조하세요.

시작하기 전에

이 문서의 코드를 실행하기 전에 다음 기본 요건을 완료하세요.

결제가 사용 설정된 Google Cloud 프로젝트를 만들거나 선택합니다.
프로젝트에 Cloud Storage 버킷을 만듭니다.
애플리케이션 기본 사용자 인증 정보를 구성하고 주 구성원에게 버킷에 객체를 만들 수 있는 권한이 있는 Identity and Access Management(IAM) 역할(예: 스토리지 객체 사용자(roles/storage.objectUser))을 부여합니다.

종속 항목

이 문서에서 사용되는 Cloud Storage 클라이언트 라이브러리와 지원 크레이트를 Cargo.toml 파일에 추가합니다.

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

큐 수신기를 스트리밍 소스로 변환

데이터를 생성하는 태스크를 페이로드를 업로드하는 태스크에서 분리하려면 비동기 큐로 연결합니다. 이 문서에서는 Tokio mpsc 채널을 사용하지만 비동기 런타임과 통합되는 모든 채널 또는 큐를 사용할 수 있습니다.

채널 수신기를 StreamingSource 구현에 맞게 조정하려면 다음 단계를 완료하세요.

채널 Receiver를 래핑하는 구조체를 정의합니다.

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

next를 호출할 때마다 수신기의 다음 청크를 기다리도록 래퍼 구조체에 StreamingSource 특성을 구현합니다.

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

큐에서 Cloud Storage로 데이터 스트리밍

StreamingSource 특성을 구현한 후 백그라운드 작업에서 객체 업로드를 생성하고 프로듀서 작업에서 채널을 통해 청크를 전송할 수 있습니다. 채널 수신기는 Seek를 구현하지 않는 단일 패스 데이터 소스이므로 send_unbuffered 대신 send_buffered를 사용하여 업로드를 시작합니다.

projects/_/buckets/BUCKET_ID 및 object_name 형식의 bucket_name를 허용하는 async 함수 내에서 다음 단계를 완료합니다.

Storage 클라이언트를 초기화합니다.

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

제한된 mpsc 채널을 만들어 sender 및 receiver 핸들을 가져옵니다.
```
let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);
```
QueueSource(receiver)를 페이로드 소스로 사용하여 객체 업로드를 구성합니다. 아직 send_buffered Future를 기다리지 마세요.
```
let upload = client
    .write_object(bucket_name, object_name, QueueSource(receiver))
    .send_buffered();
```
프로듀서가 데이터를 생성하는 동안 업로드를 구동하는 백그라운드 태스크를 생성합니다.
```
let task = tokio::spawn(upload);
```

프로듀서 태스크에서 데이터 청크를 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?;
}

모든 데이터를 전송한 후 sender 핸들을 삭제하여 채널을 닫고 객체 스트림의 끝을 알립니다.
```
drop(sender);
```
백그라운드 작업 핸들을 기다려 작업과 업로드 요청이 모두 성공했는지 확인하고 업로드된 Object 메타데이터를 가져옵니다.
```
let object = task.await??;
println!("object successfully uploaded {object:?}");
```

삭제

이 문서에서 사용한 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 완료 후 업로드한 테스트 객체를 삭제하세요.

코드 완성

다음 예시에서는 이전의 모든 단계를 완전한 함수로 결합합니다.

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