gcs-rsync 0.4.8

rsync support for gcs with higher perf than gsutil rsync
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

use futures::{Stream, StreamExt, TryStream, TryStreamExt};

use crate::oauth2::token::TokenGenerator;

use super::{
    client::StorageClient,
    resources::object::{ObjectMetadata, Objects},
    Bucket, StorageResult, {Object, ObjectsListRequest, PartialObject},
};

pub struct ObjectClient {
    storage_client: StorageClient,
}

impl ObjectClient {
    pub async fn new(token_generator: Box<dyn TokenGenerator>) -> StorageResult<Self> {
        Ok(Self {
            storage_client: StorageClient::new(token_generator).await?,
        })
    }

    pub fn no_auth() -> Self {
        Self {
            storage_client: StorageClient::no_auth(),
        }
    }

    pub async fn get(&self, o: &Object, fields: &str) -> StorageResult<PartialObject> {
        let url = o.url();
        self.storage_client
            .get_as_json(url.as_str(), &[("fields", fields)])
            .await
    }

    pub async fn delete(&self, o: &Object) -> StorageResult<String> {
        let url = o.url();
        self.storage_client.delete(&url).await?;
        super::StorageResult::Ok(url)
    }

    pub async fn download(
        &self,
        o: &Object,
    ) -> StorageResult<impl Stream<Item = StorageResult<bytes::Bytes>>> {
        let url = o.url();
        self.storage_client
            .get_as_stream(&url, &[("alt", "media")])
            .await
    }

    pub async fn upload<S>(&self, o: &Object, stream: S) -> StorageResult<()>
    where
        S: futures::TryStream + Send + Sync + 'static,
        S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
        bytes::Bytes: From<S::Ok>,
    {
        let url = o.upload_url("media");
        self.storage_client.post(&url, stream).await?;
        super::StorageResult::Ok(())
    }

    pub async fn upload_with_metadata<S>(
        &self,
        m: &ObjectMetadata,
        o: &Object,
        stream: S,
    ) -> StorageResult<()>
    where
        S: TryStream<Ok = bytes::Bytes> + Send + Sync + 'static,
        S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync,
    {
        let url = o.upload_url("multipart");
        self.storage_client.post_multipart(&url, m, stream).await?;
        super::StorageResult::Ok(())
    }

    fn list_url(bucket: &str) -> String {
        format!("{}/o", Bucket::new(bucket).url())
    }

    pub async fn is_valid(&self, bucket: &str, prefix: &str) -> StorageResult<()> {
        // For public bucket, the json api for bucket does not work to check if a bucket exists (surely for security reason).
        // Only listing operation works. Listing the first page is quick.
        let objects_list_request = ObjectsListRequest {
            prefix: Some(prefix.to_owned()),
            fields: Some("items(name)".to_owned()),
            max_results: Some(1),
            ..Default::default()
        };

        let url = Self::list_url(bucket);
        self.storage_client
            .get_as_json(&url, &objects_list_request)
            .await
            .map(|_: Objects| ())
    }

    pub async fn list(
        &self,
        bucket: &str,
        objects_list_request: &ObjectsListRequest,
    ) -> impl Stream<Item = StorageResult<PartialObject>> + '_ {
        let objects_list_request = objects_list_request.to_owned();
        let url = Self::list_url(bucket);
        futures::stream::try_unfold(
            (Some(objects_list_request), url),
            move |(state, url)| async move {
                match state {
                    None => Ok(None),
                    Some(state) => {
                        let objects: Objects =
                            self.storage_client.get_as_json(&url, &state).await?;
                        let items = futures::stream::iter(objects.items).map(Ok);
                        match objects.next_page_token {
                            None => Ok(Some((items, (None, url)))),
                            Some(next_token) => {
                                let new_state = ObjectsListRequest {
                                    page_token: Some(next_token),
                                    ..state
                                };
                                Ok(Some((items, (Some(new_state), url))))
                            }
                        }
                    }
                }
            },
        )
        .try_flatten()
    }
}