gcs_rsync/gcp/storage/
object.rs

1use futures::{Stream, StreamExt, TryStream, TryStreamExt};
2
3use crate::oauth2::token::TokenGenerator;
4
5use super::{
6    client::StorageClient,
7    resources::object::{ObjectMetadata, Objects},
8    Bucket, StorageResult, {Object, ObjectsListRequest, PartialObject},
9};
10
11pub struct ObjectClient {
12    storage_client: StorageClient,
13}
14
15impl ObjectClient {
16    pub async fn new(token_generator: Box<dyn TokenGenerator>) -> StorageResult<Self> {
17        Ok(Self {
18            storage_client: StorageClient::new(token_generator).await?,
19        })
20    }
21
22    pub fn no_auth() -> Self {
23        Self {
24            storage_client: StorageClient::no_auth(),
25        }
26    }
27
28    pub async fn get(&self, o: &Object, fields: &str) -> StorageResult<PartialObject> {
29        let url = o.url();
30        self.storage_client
31            .get_as_json(url.as_str(), &[("fields", fields)])
32            .await
33    }
34
35    pub async fn delete(&self, o: &Object) -> StorageResult<String> {
36        let url = o.url();
37        self.storage_client.delete(&url).await?;
38        super::StorageResult::Ok(url)
39    }
40
41    pub async fn download(
42        &self,
43        o: &Object,
44    ) -> StorageResult<impl Stream<Item = StorageResult<bytes::Bytes>>> {
45        let url = o.url();
46        self.storage_client
47            .get_as_stream(&url, &[("alt", "media")])
48            .await
49    }
50
51    pub async fn upload<S>(&self, o: &Object, stream: S) -> StorageResult<()>
52    where
53        S: futures::TryStream + Send + Sync + 'static,
54        S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
55        bytes::Bytes: From<S::Ok>,
56    {
57        let url = o.upload_url("media");
58        self.storage_client.post(&url, stream).await?;
59        super::StorageResult::Ok(())
60    }
61
62    pub async fn upload_with_metadata<S>(
63        &self,
64        m: &ObjectMetadata,
65        o: &Object,
66        stream: S,
67    ) -> StorageResult<()>
68    where
69        S: TryStream<Ok = bytes::Bytes> + Send + Sync + 'static,
70        S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync,
71    {
72        let url = o.upload_url("multipart");
73        self.storage_client.post_multipart(&url, m, stream).await?;
74        super::StorageResult::Ok(())
75    }
76
77    pub async fn list(
78        &self,
79        bucket: &str,
80        objects_list_request: &ObjectsListRequest,
81    ) -> impl Stream<Item = StorageResult<PartialObject>> + '_ {
82        let objects_list_request = objects_list_request.to_owned();
83        let url = Bucket::new(bucket).url();
84        futures::stream::try_unfold(
85            (Some(objects_list_request), url),
86            move |(state, url)| async move {
87                match state {
88                    None => Ok(None),
89                    Some(state) => {
90                        let objects: Objects =
91                            self.storage_client.get_as_json(&url, &state).await?;
92                        let items = futures::stream::iter(objects.items).map(Ok);
93                        match objects.next_page_token {
94                            None => Ok(Some((items, (None, url)))),
95                            Some(next_token) => {
96                                let new_state = ObjectsListRequest {
97                                    page_token: Some(next_token),
98                                    ..state
99                                };
100                                Ok(Some((items, (Some(new_state), url))))
101                            }
102                        }
103                    }
104                }
105            },
106        )
107        .try_flatten()
108    }
109}
gcs_rsync/gcp/storage/object.rs

gcs_rsync/gcp/storage/
object.rs
