cloud_storage/resources/
object.rs

1pub use crate::resources::bucket::Owner;
2use crate::resources::object_access_control::ObjectAccessControl;
3use futures_util::Stream;
4#[cfg(feature = "global-client")]
5use futures_util::TryStream;
6use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
7use std::collections::HashMap;
8
9/// A resource representing a file in Google Cloud Storage.
10#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
11#[serde(rename_all = "camelCase")]
12pub struct Object {
13    /// The kind of item this is. For objects, this is always `storage#object`.
14    pub kind: String,
15    /// The ID of the object, including the bucket name, object name, and generation number.
16    pub id: String,
17    /// The link to this object.
18    pub self_link: String,
19    /// The name of the object. Required if not specified by URL parameter.
20    pub name: String,
21    /// The name of the bucket containing this object.
22    pub bucket: String,
23    /// The content generation of this object. Used for object versioning.
24    #[serde(deserialize_with = "crate::from_str")]
25    pub generation: i64,
26    /// The version of the metadata for this object at this generation. Used for preconditions and
27    /// for detecting changes in metadata. A metageneration number is only meaningful in the context
28    /// of a particular generation of a particular object.
29    #[serde(deserialize_with = "crate::from_str")]
30    pub metageneration: i64,
31    /// Content-Type of the object data. If an object is stored without a Content-Type, it is served
32    /// as application/octet-stream.
33    pub content_type: Option<String>,
34    /// The creation time of the object in RFC 3339 format.
35    pub time_created: chrono::DateTime<chrono::Utc>,
36    /// The modification time of the object metadata in RFC 3339 format.
37    pub updated: chrono::DateTime<chrono::Utc>,
38    /// The deletion time of the object in RFC 3339 format. Returned if and only if this version of
39    /// the object is no longer a live version, but remains in the bucket as a noncurrent version.
40    pub time_deleted: Option<chrono::DateTime<chrono::Utc>>,
41    /// Whether or not the object is subject to a temporary hold.
42    pub temporary_hold: Option<bool>,
43    /// Whether or not the object is subject to an event-based hold.
44    pub event_based_hold: Option<bool>,
45    /// The earliest time that the object can be deleted, based on a bucket's retention policy, in
46    /// RFC 3339 format.
47    pub retention_expiration_time: Option<chrono::DateTime<chrono::Utc>>,
48    /// Storage class of the object.
49    pub storage_class: String,
50    /// The time at which the object's storage class was last changed. When the object is initially
51    /// created, it will be set to timeCreated.
52    pub time_storage_class_updated: chrono::DateTime<chrono::Utc>,
53    /// Content-Length of the data in bytes.
54    #[serde(deserialize_with = "crate::from_str")]
55    pub size: u64,
56    /// MD5 hash of the data; encoded using base64. For more information about using the MD5 hash,
57    /// see Hashes and ETags: Best Practices.
58    pub md5_hash: Option<String>,
59    /// Media download link.
60    pub media_link: String,
61    /// Content-Encoding of the object data.
62    pub content_encoding: Option<String>,
63    /// Content-Disposition of the object data.
64    pub content_disposition: Option<String>,
65    /// Content-Language of the object data.
66    pub content_language: Option<String>,
67    /// Cache-Control directive for the object data. If omitted, and the object is accessible to all
68    /// anonymous users, the default will be public, max-age=3600.
69    pub cache_control: Option<String>,
70    /// User-provided metadata, in key/value pairs.
71    pub metadata: Option<std::collections::HashMap<String, String>>,
72    /// Access controls on the object, containing one or more objectAccessControls Resources. If
73    /// iamConfiguration.uniformBucketLevelAccess.enabled is set to true, this field is omitted in
74    /// responses, and requests that specify this field fail.
75    pub acl: Option<Vec<ObjectAccessControl>>,
76    /// The owner of the object. This will always be the uploader of the object. If
77    /// `iamConfiguration.uniformBucketLevelAccess.enabled` is set to true, this field does not
78    /// apply, and is omitted in responses.
79    pub owner: Option<Owner>,
80    /// CRC32c checksum, as described in RFC 4960, Appendix B; encoded using base64 in big-endian
81    /// byte order. For more information about using the CRC32c checksum, see Hashes and ETags: Best
82    /// Practices.
83    pub crc32c: String,
84    /// Number of underlying components that make up a composite object. Components are accumulated
85    /// by compose operations, counting 1 for each non-composite source object and componentCount
86    /// for each composite source object. Note: componentCount is included in the metadata for
87    /// composite objects only.
88    #[serde(default, deserialize_with = "crate::from_str_opt")]
89    pub component_count: Option<i32>,
90    /// HTTP 1.1 Entity tag for the object.
91    pub etag: String,
92    /// Metadata of customer-supplied encryption key, if the object is encrypted by such a key.
93    pub customer_encryption: Option<CustomerEncrypton>,
94    /// Cloud KMS Key used to encrypt this object, if the object is encrypted by such a key.
95    pub kms_key_name: Option<String>,
96}
97
98/// Contains data about how a user might encrypt their files in Google Cloud Storage.
99#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
100#[serde(rename_all = "camelCase")]
101pub struct CustomerEncrypton {
102    /// The encryption algorithm.
103    pub encryption_algorithm: String,
104    /// SHA256 hash value of the encryption key.
105    pub key_sha256: String,
106}
107
108/// The request that is supplied to perform `Object::compose`.
109#[derive(Debug, PartialEq, serde::Serialize)]
110#[serde(rename_all = "camelCase")]
111pub struct ComposeRequest {
112    /// The kind of item this is. Will always be `storage#composeRequest`.
113    pub kind: String,
114    /// The list of source objects that will be concatenated into a single object.
115    pub source_objects: Vec<SourceObject>,
116    /// Properties of the resulting object.
117    pub destination: Option<Object>,
118}
119
120/// A SourceObject represents one of the objects that is to be composed.
121#[derive(Debug, PartialEq, serde::Serialize)]
122#[serde(rename_all = "camelCase")]
123pub struct SourceObject {
124    /// The source object's name. All source objects must have the same storage class and reside in
125    /// the same bucket.
126    pub name: String,
127    /// The generation of this object to use as the source.
128    pub generation: Option<i64>,
129    /// Conditions that must be met for this operation to execute.
130    pub object_preconditions: Option<ObjectPrecondition>,
131}
132
133/// Allows conditional copying of this file.
134#[derive(Debug, PartialEq, serde::Serialize)]
135#[serde(rename_all = "camelCase")]
136pub struct ObjectPrecondition {
137    /// Only perform the composition if the generation of the source object that would be used
138    /// matches this value. If this value and a generation are both specified, they must be the same
139    /// value or the call will fail.
140    pub if_generation_match: i64,
141}
142
143/// The request that is supplied to perform `Object::list`.
144/// See [the Google Cloud Storage API
145/// reference](https://cloud.google.com/storage/docs/json_api/v1/objects/list)
146/// for more details.
147#[derive(Debug, PartialEq, serde::Serialize, Default, Clone)]
148#[serde(rename_all = "camelCase")]
149pub struct ListRequest {
150    /// When specified, allows the `list` to operate like a directory listing by splitting the
151    /// object location on this delimiter.
152    pub delimiter: Option<String>,
153
154    /// Filter results to objects whose names are lexicographically before `end_offset`.
155    /// If `start_offset` is also set, the objects listed have names between `start_offset`
156    /// (inclusive) and `end_offset` (exclusive).
157    pub end_offset: Option<String>,
158
159    /// If true, objects that end in exactly one instance of `delimiter` have their metadata
160    /// included in `items` in addition to the relevant part of the object name appearing in
161    /// `prefixes`.
162    pub include_trailing_delimiter: Option<bool>,
163
164    /// Maximum combined number of entries in `items` and `prefixes` to return in a single
165    /// page of responses. Because duplicate entries in `prefixes` are omitted, fewer total
166    /// results may be returned than requested. The service uses this parameter or 1,000
167    /// items, whichever is smaller.
168    pub max_results: Option<usize>,
169
170    /// A previously-returned page token representing part of the larger set of results to view.
171    /// The `page_token` is an encoded field that marks the name and generation of the last object
172    /// in the returned list. In a subsequent request using the `page_token`, items that come after
173    /// the `page_token` are shown (up to `max_results`).
174    ///
175    /// If the page token is provided, all objects starting at that page token are queried
176    pub page_token: Option<String>,
177
178    /// Filter results to include only objects whose names begin with this prefix.
179    pub prefix: Option<String>,
180
181    /// Set of properties to return. Defaults to `NoAcl`.
182    pub projection: Option<Projection>,
183
184    /// Filter results to objects whose names are lexicographically equal to or after
185    /// `start_offset`. If `end_offset` is also set, the objects listed have names between
186    /// `start_offset` (inclusive) and `end_offset` (exclusive).
187    pub start_offset: Option<String>,
188
189    /// If true, lists all versions of an object as distinct results in order of increasing
190    /// generation number. The default value for versions is false. For more information, see
191    /// Object Versioning.
192    pub versions: Option<bool>,
193}
194
195/// Acceptable values of `projection` properties to return from `Object::list` requests.
196#[derive(Debug, PartialEq, serde::Serialize, Clone)]
197#[serde(rename_all = "camelCase")]
198pub enum Projection {
199    /// Include all properties.
200    Full,
201    /// Omit the owner, acl property.
202    NoAcl,
203}
204
205/// Response from `Object::list`.
206#[derive(Debug, serde::Deserialize, Default)]
207#[serde(rename_all = "camelCase")]
208pub struct ObjectList {
209    /// The kind of item this is. For lists of objects, this is always `storage#objects`.
210    pub kind: String,
211
212    /// The list of objects, ordered lexicographically by name.
213    #[serde(default = "Vec::new")]
214    pub items: Vec<Object>,
215
216    /// Object name prefixes for objects that matched the listing request but were excluded
217    /// from `items` because of a delimiter. Values in this list are object names up to and
218    /// including the requested delimiter. Duplicate entries are omitted from this list.
219    #[serde(default = "Vec::new")]
220    pub prefixes: Vec<String>,
221
222    /// The continuation token, included only if there are more items to return. Provide
223    /// this value as the `page_token` of a subsequent request in order to return the next
224    /// page of results.
225    pub next_page_token: Option<String>,
226}
227
228#[derive(Debug, serde::Deserialize)]
229#[serde(rename_all = "camelCase")]
230#[allow(dead_code)]
231pub(crate) struct RewriteResponse {
232    kind: String,
233    total_bytes_rewritten: String,
234    object_size: String,
235    done: bool,
236    pub(crate) resource: Object,
237}
238
239impl Object {
240    /// Create a new object.
241    /// Upload a file as that is loaded in memory to google cloud storage, where it will be
242    /// interpreted according to the mime type you specified.
243    /// ## Example
244    /// ```rust,no_run
245    /// # #[tokio::main]
246    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
247    /// # fn read_cute_cat(_in: &str) -> Vec<u8> { vec![0, 1] }
248    /// use cloud_storage::Object;
249    ///
250    /// let file: Vec<u8> = read_cute_cat("cat.png");
251    /// Object::create("cat-photos", file, "recently read cat.png", "image/png").await?;
252    /// # Ok(())
253    /// # }
254    /// ```
255    #[cfg(feature = "global-client")]
256    pub async fn create(
257        bucket: &str,
258        file: Vec<u8>,
259        filename: &str,
260        mime_type: &str,
261    ) -> crate::Result<Self> {
262        crate::CLOUD_CLIENT
263            .object()
264            .create(bucket, file, filename, mime_type)
265            .await
266    }
267
268    /// The synchronous equivalent of `Object::create`.
269    ///
270    /// ### Features
271    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
272    #[cfg(all(feature = "global-client", feature = "sync"))]
273    pub fn create_sync(
274        bucket: &str,
275        file: Vec<u8>,
276        filename: &str,
277        mime_type: &str,
278    ) -> crate::Result<Self> {
279        crate::runtime()?.block_on(Self::create(bucket, file, filename, mime_type))
280    }
281
282    /// Create a new object. This works in the same way as `Object::create`, except it does not need
283    /// to load the entire file in ram.
284    /// ## Example
285    /// ```rust,no_run
286    /// # #[tokio::main]
287    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
288    /// use cloud_storage::Object;
289    ///
290    /// let file = reqwest::Client::new()
291    ///     .get("https://my_domain.rs/nice_cat_photo.png")
292    ///     .send()
293    ///     .await?
294    ///     .bytes_stream();
295    /// Object::create_streamed("cat-photos", file, 10, "recently read cat.png", "image/png").await?;
296    /// # Ok(())
297    /// # }
298    /// ```
299    #[cfg(feature = "global-client")]
300    pub async fn create_streamed<S>(
301        bucket: &str,
302        stream: S,
303        length: impl Into<Option<u64>>,
304        filename: &str,
305        mime_type: &str,
306    ) -> crate::Result<Self>
307    where
308        S: TryStream + Send + Sync + 'static,
309        S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
310        bytes::Bytes: From<S::Ok>,
311    {
312        crate::CLOUD_CLIENT
313            .object()
314            .create_streamed(bucket, stream, length, filename, mime_type)
315            .await
316    }
317
318    /// The synchronous equivalent of `Object::create_streamed`.
319    ///
320    /// ### Features
321    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
322    #[cfg(all(feature = "global-client", feature = "sync"))]
323    pub fn create_streamed_sync<R: std::io::Read + Send + 'static>(
324        bucket: &str,
325        mut file: R,
326        length: impl Into<Option<u64>>,
327        filename: &str,
328        mime_type: &str,
329    ) -> crate::Result<Self> {
330        let mut buffer = Vec::new();
331        file.read_to_end(&mut buffer)
332            .map_err(|e| crate::Error::Other(e.to_string()))?;
333
334        let stream = futures_util::stream::once(async { Ok::<_, crate::Error>(buffer) });
335
336        crate::runtime()?.block_on(Self::create_streamed(
337            bucket, stream, length, filename, mime_type,
338        ))
339    }
340
341    /// Obtain a list of objects within this Bucket. This function will repeatedly query Google and
342    /// merge the responses into one. Google responds with 1000 Objects at a time, so if you want to
343    /// make sure only one http call is performed, make sure to set `list_request.max_results` to
344    /// 1000.
345    /// ### Example
346    /// ```no_run
347    /// # #[tokio::main]
348    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
349    /// use cloud_storage::{Object, ListRequest};
350    ///
351    /// let all_objects = Object::list("my_bucket", ListRequest::default()).await?;
352    /// # Ok(())
353    /// # }
354    /// ```
355    #[cfg(feature = "global-client")]
356    pub async fn list(
357        bucket: &str,
358        list_request: ListRequest,
359    ) -> crate::Result<impl Stream<Item = crate::Result<ObjectList>> + '_> {
360        crate::CLOUD_CLIENT
361            .object()
362            .list(bucket, list_request)
363            .await
364    }
365
366    /// The synchronous equivalent of `Object::list`.
367    ///
368    /// ### Features
369    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
370    #[cfg(all(feature = "global-client", feature = "sync"))]
371    pub fn list_sync(bucket: &str, list_request: ListRequest) -> crate::Result<Vec<ObjectList>> {
372        use futures_util::TryStreamExt;
373
374        let rt = crate::runtime()?;
375        let listed = rt.block_on(Self::list(bucket, list_request))?;
376        rt.block_on(listed.try_collect())
377    }
378
379    /// Obtains a single object with the specified name in the specified bucket.
380    /// ### Example
381    /// ```no_run
382    /// # #[tokio::main]
383    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
384    /// use cloud_storage::Object;
385    ///
386    /// let object = Object::read("my_bucket", "path/to/my/file.png").await?;
387    /// # Ok(())
388    /// # }
389    /// ```
390    #[cfg(feature = "global-client")]
391    pub async fn read(bucket: &str, file_name: &str) -> crate::Result<Self> {
392        crate::CLOUD_CLIENT.object().read(bucket, file_name).await
393    }
394
395    /// The synchronous equivalent of `Object::read`.
396    ///
397    /// ### Features
398    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
399    #[cfg(all(feature = "global-client", feature = "sync"))]
400    pub fn read_sync(bucket: &str, file_name: &str) -> crate::Result<Self> {
401        crate::runtime()?.block_on(Self::read(bucket, file_name))
402    }
403
404    /// Download the content of the object with the specified name in the specified bucket.
405    /// ### Example
406    /// ```no_run
407    /// # #[tokio::main]
408    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
409    /// use cloud_storage::Object;
410    ///
411    /// let bytes = Object::download("my_bucket", "path/to/my/file.png").await?;
412    /// # Ok(())
413    /// # }
414    /// ```
415    #[cfg(feature = "global-client")]
416    pub async fn download(bucket: &str, file_name: &str) -> crate::Result<Vec<u8>> {
417        crate::CLOUD_CLIENT
418            .object()
419            .download(bucket, file_name)
420            .await
421    }
422
423    /// The synchronous equivalent of `Object::download`.
424    ///
425    /// ### Features
426    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
427    #[cfg(all(feature = "global-client", feature = "sync"))]
428    pub fn download_sync(bucket: &str, file_name: &str) -> crate::Result<Vec<u8>> {
429        crate::runtime()?.block_on(Self::download(bucket, file_name))
430    }
431
432    /// Download the content of the object with the specified name in the specified bucket, without
433    /// allocating the whole file into a vector.
434    /// ### Example
435    /// ```no_run
436    /// # #[tokio::main]
437    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
438    /// use cloud_storage::Object;
439    /// use futures_util::stream::StreamExt;
440    /// use std::fs::File;
441    /// use std::io::{BufWriter, Write};
442    ///
443    /// let mut stream = Object::download_streamed("my_bucket", "path/to/my/file.png").await?;
444    /// let mut file = BufWriter::new(File::create("file.png").unwrap());
445    /// while let Some(byte) = stream.next().await {
446    ///     file.write_all(&[byte.unwrap()]).unwrap();
447    /// }
448    /// # Ok(())
449    /// # }
450    /// ```
451    #[cfg(feature = "global-client")]
452    pub async fn download_streamed(
453        bucket: &str,
454        file_name: &str,
455    ) -> crate::Result<impl Stream<Item = crate::Result<u8>> + Unpin> {
456        crate::CLOUD_CLIENT
457            .object()
458            .download_streamed(bucket, file_name)
459            .await
460    }
461
462    /// Obtains a single object with the specified name in the specified bucket.
463    /// ### Example
464    /// ```no_run
465    /// # #[tokio::main]
466    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
467    /// use cloud_storage::Object;
468    ///
469    /// let mut object = Object::read("my_bucket", "path/to/my/file.png").await?;
470    /// object.content_type = Some("application/xml".to_string());
471    /// object.update().await?;
472    /// # Ok(())
473    /// # }
474    /// ```
475    #[cfg(feature = "global-client")]
476    pub async fn update(&self) -> crate::Result<Self> {
477        crate::CLOUD_CLIENT.object().update(self).await
478    }
479
480    /// The synchronous equivalent of `Object::download`.
481    ///
482    /// ### Features
483    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
484    #[cfg(all(feature = "global-client", feature = "sync"))]
485    pub fn update_sync(&self) -> crate::Result<Self> {
486        crate::runtime()?.block_on(self.update())
487    }
488
489    /// Deletes a single object with the specified name in the specified bucket.
490    /// ### Example
491    /// ```no_run
492    /// # #[tokio::main]
493    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
494    /// use cloud_storage::Object;
495    ///
496    /// Object::delete("my_bucket", "path/to/my/file.png").await?;
497    /// # Ok(())
498    /// # }
499    /// ```
500    #[cfg(feature = "global-client")]
501    pub async fn delete(bucket: &str, file_name: &str) -> crate::Result<()> {
502        crate::CLOUD_CLIENT.object().delete(bucket, file_name).await
503    }
504
505    /// The synchronous equivalent of `Object::delete`.
506    ///
507    /// ### Features
508    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
509    #[cfg(all(feature = "global-client", feature = "sync"))]
510    pub fn delete_sync(bucket: &str, file_name: &str) -> crate::Result<()> {
511        crate::runtime()?.block_on(Self::delete(bucket, file_name))
512    }
513
514    /// Obtains a single object with the specified name in the specified bucket.
515    /// ### Example
516    /// ```no_run
517    /// # #[tokio::main]
518    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
519    /// use cloud_storage::object::{Object, ComposeRequest, SourceObject};
520    ///
521    /// let obj1 = Object::read("my_bucket", "file1").await?;
522    /// let obj2 = Object::read("my_bucket", "file2").await?;
523    /// let compose_request = ComposeRequest {
524    ///     kind: "storage#composeRequest".to_string(),
525    ///     source_objects: vec![
526    ///         SourceObject {
527    ///             name: obj1.name.clone(),
528    ///             generation: None,
529    ///             object_preconditions: None,
530    ///         },
531    ///         SourceObject {
532    ///             name: obj2.name.clone(),
533    ///             generation: None,
534    ///             object_preconditions: None,
535    ///         },
536    ///     ],
537    ///     destination: None,
538    /// };
539    /// let obj3 = Object::compose("my_bucket", &compose_request, "test-concatted-file").await?;
540    /// // obj3 is now a file with the content of obj1 and obj2 concatted together.
541    /// # Ok(())
542    /// # }
543    /// ```
544    #[cfg(feature = "global-client")]
545    pub async fn compose(
546        bucket: &str,
547        req: &ComposeRequest,
548        destination_object: &str,
549    ) -> crate::Result<Self> {
550        crate::CLOUD_CLIENT
551            .object()
552            .compose(bucket, req, destination_object)
553            .await
554    }
555
556    /// The synchronous equivalent of `Object::compose`.
557    ///
558    /// ### Features
559    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
560    #[cfg(all(feature = "global-client", feature = "sync"))]
561    pub fn compose_sync(
562        bucket: &str,
563        req: &ComposeRequest,
564        destination_object: &str,
565    ) -> crate::Result<Self> {
566        crate::runtime()?.block_on(Self::compose(bucket, req, destination_object))
567    }
568
569    /// Copy this object to the target bucket and path
570    /// ### Example
571    /// ```no_run
572    /// # #[tokio::main]
573    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
574    /// use cloud_storage::object::{Object, ComposeRequest};
575    ///
576    /// let obj1 = Object::read("my_bucket", "file1").await?;
577    /// let obj2 = obj1.copy("my_other_bucket", "file2").await?;
578    /// // obj2 is now a copy of obj1.
579    /// # Ok(())
580    /// # }
581    /// ```
582    #[cfg(feature = "global-client")]
583    pub async fn copy(&self, destination_bucket: &str, path: &str) -> crate::Result<Self> {
584        crate::CLOUD_CLIENT
585            .object()
586            .copy(self, destination_bucket, path)
587            .await
588    }
589
590    /// The synchronous equivalent of `Object::copy`.
591    ///
592    /// ### Features
593    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
594    #[cfg(all(feature = "global-client", feature = "sync"))]
595    pub fn copy_sync(&self, destination_bucket: &str, path: &str) -> crate::Result<Self> {
596        crate::runtime()?.block_on(self.copy(destination_bucket, path))
597    }
598
599    /// Moves a file from the current location to the target bucket and path.
600    ///
601    /// ## Limitations
602    /// This function does not yet support rewriting objects to another
603    /// * Geographical Location,
604    /// * Encryption,
605    /// * Storage class.
606    /// These limitations mean that for now, the rewrite and the copy methods do the same thing.
607    /// ### Example
608    /// ```no_run
609    /// # #[tokio::main]
610    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
611    /// use cloud_storage::object::Object;
612    ///
613    /// let obj1 = Object::read("my_bucket", "file1").await?;
614    /// let obj2 = obj1.rewrite("my_other_bucket", "file2").await?;
615    /// // obj2 is now a copy of obj1.
616    /// # Ok(())
617    /// # }
618    /// ```
619    #[cfg(feature = "global-client")]
620    pub async fn rewrite(&self, destination_bucket: &str, path: &str) -> crate::Result<Self> {
621        crate::CLOUD_CLIENT
622            .object()
623            .rewrite(self, destination_bucket, path)
624            .await
625    }
626
627    /// The synchronous equivalent of `Object::rewrite`.
628    ///
629    /// ### Features
630    /// This function requires that the feature flag `sync` is enabled in `Cargo.toml`.
631    #[cfg(all(feature = "global-client", feature = "sync"))]
632    pub fn rewrite_sync(&self, destination_bucket: &str, path: &str) -> crate::Result<Self> {
633        crate::runtime()?.block_on(self.rewrite(destination_bucket, path))
634    }
635
636    /// Creates a [Signed Url](https://cloud.google.com/storage/docs/access-control/signed-urls)
637    /// which is valid for `duration` seconds, and lets the posessor download the file contents
638    /// without any authentication.
639    /// ### Example
640    /// ```no_run
641    /// # #[tokio::main]
642    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
643    /// use cloud_storage::{Client, object::{Object, ComposeRequest}};
644    ///
645    /// let client = Client::default();
646    /// let obj1 = client.object().read("my_bucket", "file1").await?;
647    /// let url = obj1.download_url(50)?;
648    /// // url is now a url to which an unauthenticated user can make a request to download a file
649    /// // for 50 seconds.
650    /// # Ok(())
651    /// # }
652    /// ```
653    pub fn download_url(&self, duration: u32) -> crate::Result<String> {
654        self.sign(&self.name, duration, "GET", None, &HashMap::new())
655    }
656
657    /// Creates a [Signed Url](https://cloud.google.com/storage/docs/access-control/signed-urls)
658    /// which is valid for `duration` seconds, and lets the posessor download the file contents
659    /// without any authentication.
660    /// ### Example
661    /// ```no_run
662    /// # #[tokio::main]
663    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
664    /// use cloud_storage::{Client, object::{Object, ComposeRequest}};
665    ///
666    /// let client = Client::default();
667    /// let obj1 = client.object().read("my_bucket", "file1").await?;
668    /// let url = obj1.download_url(50)?;
669    /// // url is now a url to which an unauthenticated user can make a request to download a file
670    /// // for 50 seconds.
671    /// # Ok(())
672    /// # }
673    /// ```
674    pub fn download_url_with(
675        &self,
676        duration: u32,
677        opts: crate::DownloadOptions,
678    ) -> crate::Result<String> {
679        self.sign(
680            &self.name,
681            duration,
682            "GET",
683            opts.content_disposition,
684            &HashMap::new(),
685        )
686    }
687
688    /// Creates a [Signed Url](https://cloud.google.com/storage/docs/access-control/signed-urls)
689    /// which is valid for `duration` seconds, and lets the posessor upload data to a blob
690    /// without any authentication.
691    /// ### Example
692    /// ```no_run
693    /// # #[tokio::main]
694    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
695    /// use cloud_storage::{Client, object::{Object, ComposeRequest}};
696    ///
697    /// let client = Client::default();
698    /// let obj1 = client.object().read("my_bucket", "file1").await?;
699    /// let url = obj1.upload_url(50)?;
700    /// // url is now a url to which an unauthenticated user can make a PUT request to upload a file
701    /// // for 50 seconds.
702    /// # Ok(())
703    /// # }
704    /// ```
705    pub fn upload_url(&self, duration: u32) -> crate::Result<String> {
706        self.sign(&self.name, duration, "PUT", None, &HashMap::new())
707    }
708
709    /// Creates a [Signed Url](https://cloud.google.com/storage/docs/access-control/signed-urls)
710    /// which is valid for `duration` seconds, and lets the posessor upload data and custom metadata
711    /// to a blob without any authentication.
712    /// ### Example
713    /// ```no_run
714    /// # #[tokio::main]
715    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
716    /// use cloud_storage::{Client, object::{Object, ComposeRequest}};
717    /// use std::collections::HashMap;
718    ///
719    /// let client = Client::default();
720    /// let obj1 = client.object().read("my_bucket", "file1").await?;
721    /// let mut custom_metadata = HashMap::new();
722    /// custom_metadata.insert(String::from("field"), String::from("value"));
723    /// let (url, headers) = obj1.upload_url_with(50, custom_metadata)?;
724    /// // url is now a url to which an unauthenticated user can make a PUT request to upload a file
725    /// // for 50 seconds. Note that the user must also include the returned headers in the PUT request
726    /// # Ok(())
727    /// # }
728    /// ```
729    pub fn upload_url_with(
730        &self,
731        duration: u32,
732        custom_metadata: HashMap<String, String>,
733    ) -> crate::Result<(String, HashMap<String, String>)> {
734        let url = self.sign(&self.name, duration, "PUT", None, &custom_metadata)?;
735        let mut headers = HashMap::new();
736        for (k, v) in custom_metadata.iter() {
737            headers.insert(format!("x-goog-meta-{}", k.to_string()), v.to_string());
738        }
739        Ok((url, headers))
740    }
741
742    // /// Creates a [Signed Url](https://cloud.google.com/storage/docs/access-control/signed-urls)
743    // /// which is valid for `duration` seconds, and lets the posessor upload new file contents.
744    // /// without any authentication.
745    // pub fn upload_url(&self, duration: u32) -> crate::Result<String> {
746    //     self.sign(&self.name, duration, "POST")
747    // }
748
749    #[inline(always)]
750    fn sign(
751        &self,
752        file_path: &str,
753        duration: u32,
754        http_verb: &str,
755        content_disposition: Option<String>,
756        custom_metadata: &HashMap<String, String>,
757    ) -> crate::Result<String> {
758        if duration > 604800 {
759            let msg = format!(
760                "duration may not be greater than 604800, but was {}",
761                duration
762            );
763            return Err(crate::Error::Other(msg));
764        }
765
766        // 0 Sort and construct the canonical headers
767        let mut headers = vec![("host".to_string(), "storage.googleapis.com".to_string())];
768        // Add custom metadata headers, guaranteed unique by HashMap input
769        for (k, v) in custom_metadata.iter() {
770            headers.push((format!("x-goog-meta-{}", k.to_string()), v.to_string()));
771        }
772        headers.sort_unstable_by(|(k1, _), (k2, _)| k1.cmp(k2));
773        let canonical_headers: String = headers
774            .iter()
775            .map(|(k, v)| format!("{}:{}", k.to_lowercase(), v.to_lowercase()))
776            .collect::<Vec<String>>()
777            .join("\n");
778        let signed_headers = headers
779            .iter()
780            .map(|(k, _)| k.to_lowercase())
781            .collect::<Vec<String>>()
782            .join(";");
783
784        // 1 construct the canonical request
785        let issue_date = chrono::Utc::now();
786        let file_path = self.path_to_resource(file_path);
787        let query_string = Self::get_canonical_query_string(
788            &issue_date,
789            duration,
790            &signed_headers,
791            content_disposition,
792        );
793        let canonical_request = self.get_canonical_request(
794            &file_path,
795            &query_string,
796            http_verb,
797            &canonical_headers,
798            &signed_headers,
799        );
800
801        // 2 get hex encoded SHA256 hash the canonical request
802        let hex_hash = hex::encode(crypto::sha256(canonical_request.as_bytes()).as_ref());
803
804        // 3 construct the string to sign
805        let string_to_sign = format!(
806            "{signing_algorithm}\n\
807            {current_datetime}\n\
808            {credential_scope}\n\
809            {hashed_canonical_request}",
810            signing_algorithm = "GOOG4-RSA-SHA256",
811            current_datetime = issue_date.format("%Y%m%dT%H%M%SZ"),
812            credential_scope = Self::get_credential_scope(&issue_date),
813            hashed_canonical_request = hex_hash,
814        );
815
816        // 4 sign the string to sign with RSA - SHA256
817        let signature = hex::encode(crypto::rsa_pkcs1_sha256(&string_to_sign)?);
818
819        // 5 construct the signed url
820        Ok(format!(
821            "https://storage.googleapis.com{path_to_resource}?\
822            {query_string}&\
823            X-Goog-Signature={request_signature}",
824            path_to_resource = file_path,
825            query_string = query_string,
826            request_signature = signature,
827        ))
828    }
829
830    #[inline(always)]
831    fn get_canonical_request(
832        &self,
833        path: &str,
834        query_string: &str,
835        http_verb: &str,
836        headers: &str,
837        signed_headers: &str,
838    ) -> String {
839        format!(
840            "{http_verb}\n\
841            {path_to_resource}\n\
842            {canonical_query_string}\n\
843            {canonical_headers}\n\
844            \n\
845            {signed_headers}\n\
846            {payload}",
847            http_verb = http_verb,
848            path_to_resource = path,
849            canonical_query_string = query_string,
850            canonical_headers = headers,
851            signed_headers = signed_headers,
852            payload = "UNSIGNED-PAYLOAD",
853        )
854    }
855
856    #[inline(always)]
857    fn get_canonical_query_string(
858        date: &chrono::DateTime<chrono::Utc>,
859        exp: u32,
860        headers: &str,
861        content_disposition: Option<String>,
862    ) -> String {
863        let credential = format!(
864            "{authorizer}/{scope}",
865            authorizer = crate::SERVICE_ACCOUNT.client_email,
866            scope = Self::get_credential_scope(date),
867        );
868        let mut s = format!(
869            "X-Goog-Algorithm={algo}&\
870            X-Goog-Credential={cred}&\
871            X-Goog-Date={date}&\
872            X-Goog-Expires={exp}&\
873            X-Goog-SignedHeaders={signed}",
874            algo = "GOOG4-RSA-SHA256",
875            cred = percent_encode(&credential),
876            date = date.format("%Y%m%dT%H%M%SZ"),
877            exp = exp,
878            signed = percent_encode(headers),
879        );
880        if let Some(cd) = content_disposition {
881            s.push_str(&format!("&response-content-disposition={}", cd));
882        }
883        s
884    }
885
886    #[inline(always)]
887    fn path_to_resource(&self, path: &str) -> String {
888        format!(
889            "/{bucket}/{file_path}",
890            bucket = self.bucket,
891            file_path = percent_encode_noslash(path),
892        )
893    }
894
895    #[inline(always)]
896    fn get_credential_scope(date: &chrono::DateTime<chrono::Utc>) -> String {
897        format!("{}/henk/storage/goog4_request", date.format("%Y%m%d"))
898    }
899}
900
901#[cfg(feature = "openssl")]
902mod openssl {
903    #[inline(always)]
904    pub fn rsa_pkcs1_sha256(message: &str) -> crate::Result<Vec<u8>> {
905        use openssl::{hash::MessageDigest, pkey::PKey, sign::Signer};
906
907        let key = PKey::private_key_from_pem(crate::SERVICE_ACCOUNT.private_key.as_bytes())?;
908        let mut signer = Signer::new(MessageDigest::sha256(), &key)?;
909        signer.update(message.as_bytes())?;
910        Ok(signer.sign_to_vec()?)
911    }
912
913    #[inline(always)]
914    pub fn sha256(bytes: &[u8]) -> impl AsRef<[u8]> {
915        openssl::sha::sha256(bytes)
916    }
917}
918
919#[cfg(feature = "ring")]
920mod ring {
921    #[cfg_attr(all(feature = "ring", feature = "openssl"), allow(dead_code))]
922    #[inline(always)]
923    pub fn rsa_pkcs1_sha256(message: &str) -> crate::Result<Vec<u8>> {
924        use ring::{
925            rand::SystemRandom,
926            signature::{RsaKeyPair, RSA_PKCS1_SHA256},
927        };
928
929        let key_pem = pem::parse(crate::SERVICE_ACCOUNT.private_key.as_bytes())?;
930        let key = RsaKeyPair::from_pkcs8(&key_pem.contents)?;
931        let rng = SystemRandom::new();
932        let mut signature = vec![0; key.public_modulus_len()];
933        key.sign(&RSA_PKCS1_SHA256, &rng, message.as_bytes(), &mut signature)?;
934        Ok(signature)
935    }
936
937    #[cfg_attr(all(feature = "ring", feature = "openssl"), allow(dead_code))]
938    #[inline(always)]
939    pub fn sha256(bytes: &[u8]) -> impl AsRef<[u8]> {
940        use ring::digest::{digest, SHA256};
941        digest(&SHA256, bytes)
942    }
943}
944
945mod crypto {
946    #[cfg(feature = "openssl")]
947    pub use super::openssl::*;
948    #[cfg(all(feature = "ring", not(feature = "openssl")))]
949    pub use super::ring::*;
950}
951
952const ENCODE_SET: &AsciiSet = &NON_ALPHANUMERIC
953    .remove(b'*')
954    .remove(b'-')
955    .remove(b'.')
956    .remove(b'_');
957
958const NOSLASH_ENCODE_SET: &AsciiSet = &ENCODE_SET.remove(b'/').remove(b'~');
959
960// We need to be able to percent encode stuff, but without touching the slashes in filenames. To
961// this end we create an implementation that does this, without touching the slashes.
962fn percent_encode_noslash(input: &str) -> String {
963    utf8_percent_encode(input, NOSLASH_ENCODE_SET).to_string()
964}
965
966pub(crate) fn percent_encode(input: &str) -> String {
967    utf8_percent_encode(input, ENCODE_SET).to_string()
968}
969
970#[cfg(all(test, feature = "global-client"))]
971mod tests {
972    use super::*;
973    use crate::Error;
974    use futures_util::{stream, StreamExt, TryStreamExt};
975
976    #[tokio::test]
977    async fn create() -> Result<(), Box<dyn std::error::Error>> {
978        let bucket = crate::read_test_bucket().await;
979        Object::create(&bucket.name, vec![0, 1], "test-create", "text/plain").await?;
980        Ok(())
981    }
982
983    #[tokio::test]
984    async fn create_streamed() -> Result<(), Box<dyn std::error::Error>> {
985        let bucket = crate::read_test_bucket().await;
986        let stream = stream::iter([0u8, 1].iter())
987            .map(Ok::<_, Box<dyn std::error::Error + Send + Sync>>)
988            .map_ok(|&b| bytes::BytesMut::from(&[b][..]));
989        Object::create_streamed(
990            &bucket.name,
991            stream,
992            2,
993            "test-create-streamed",
994            "text/plain",
995        )
996        .await?;
997        Ok(())
998    }
999
1000    #[tokio::test]
1001    async fn list() -> Result<(), Box<dyn std::error::Error>> {
1002        let test_bucket = crate::read_test_bucket().await;
1003        let _v: Vec<ObjectList> = Object::list(&test_bucket.name, ListRequest::default())
1004            .await?
1005            .try_collect()
1006            .await?;
1007        Ok(())
1008    }
1009
1010    async fn flattened_list_prefix_stream(
1011        bucket: &str,
1012        prefix: &str,
1013    ) -> Result<Vec<Object>, Box<dyn std::error::Error>> {
1014        let request = ListRequest {
1015            prefix: Some(prefix.into()),
1016            ..Default::default()
1017        };
1018
1019        Ok(Object::list(bucket, request)
1020            .await?
1021            .map_ok(|object_list| object_list.items)
1022            .try_concat()
1023            .await?)
1024    }
1025
1026    #[tokio::test]
1027    async fn list_prefix() -> Result<(), Box<dyn std::error::Error>> {
1028        let test_bucket = crate::read_test_bucket().await;
1029
1030        let prefix_names = [
1031            "test-list-prefix/1",
1032            "test-list-prefix/2",
1033            "test-list-prefix/sub/1",
1034            "test-list-prefix/sub/2",
1035        ];
1036
1037        for name in &prefix_names {
1038            Object::create(&test_bucket.name, vec![0, 1], name, "text/plain").await?;
1039        }
1040
1041        let list = flattened_list_prefix_stream(&test_bucket.name, "test-list-prefix/").await?;
1042        assert_eq!(list.len(), 4);
1043        let list = flattened_list_prefix_stream(&test_bucket.name, "test-list-prefix/sub").await?;
1044        assert_eq!(list.len(), 2);
1045        Ok(())
1046    }
1047
1048    #[tokio::test]
1049    async fn read() -> Result<(), Box<dyn std::error::Error>> {
1050        let bucket = crate::read_test_bucket().await;
1051        Object::create(&bucket.name, vec![0, 1], "test-read", "text/plain").await?;
1052        Object::read(&bucket.name, "test-read").await?;
1053        Ok(())
1054    }
1055
1056    #[tokio::test]
1057    async fn download() -> Result<(), Box<dyn std::error::Error>> {
1058        let bucket = crate::read_test_bucket().await;
1059        let content = b"hello world";
1060        Object::create(
1061            &bucket.name,
1062            content.to_vec(),
1063            "test-download",
1064            "application/octet-stream",
1065        )
1066        .await?;
1067
1068        let data = Object::download(&bucket.name, "test-download").await?;
1069        assert_eq!(data, content);
1070
1071        Ok(())
1072    }
1073
1074    #[tokio::test]
1075    async fn download_streamed() -> Result<(), Box<dyn std::error::Error>> {
1076        let bucket = crate::read_test_bucket().await;
1077        let content = b"hello world";
1078        Object::create(
1079            &bucket.name,
1080            content.to_vec(),
1081            "test-download",
1082            "application/octet-stream",
1083        )
1084        .await?;
1085
1086        let result = Object::download_streamed(&bucket.name, "test-download").await?;
1087        let data = result.try_collect::<Vec<_>>().await?;
1088        assert_eq!(data, content);
1089
1090        Ok(())
1091    }
1092
1093    #[tokio::test]
1094    async fn download_streamed_large() -> Result<(), Box<dyn std::error::Error>> {
1095        let bucket = crate::read_test_bucket().await;
1096        let content = vec![5u8; 1_000_000];
1097        Object::create(
1098            &bucket.name,
1099            content.to_vec(),
1100            "test-download-large",
1101            "application/octet-stream",
1102        )
1103        .await?;
1104
1105        let mut result = Object::download_streamed(&bucket.name, "test-download-large").await?;
1106        let mut data: Vec<u8> = Vec::new();
1107        while let Some(part) = result.next().await {
1108            data.push(part?);
1109        }
1110        assert_eq!(data, content);
1111
1112        Ok(())
1113    }
1114
1115    #[tokio::test]
1116    async fn update() -> Result<(), Box<dyn std::error::Error>> {
1117        let bucket = crate::read_test_bucket().await;
1118        let mut obj = Object::create(&bucket.name, vec![0, 1], "test-update", "text/plain").await?;
1119        obj.content_type = Some("application/xml".to_string());
1120        obj.update().await?;
1121        Ok(())
1122    }
1123
1124    #[tokio::test]
1125    async fn delete() -> Result<(), Box<dyn std::error::Error>> {
1126        let bucket = crate::read_test_bucket().await;
1127        Object::create(&bucket.name, vec![0, 1], "test-delete", "text/plain").await?;
1128
1129        Object::delete(&bucket.name, "test-delete").await?;
1130
1131        let list: Vec<_> = flattened_list_prefix_stream(&bucket.name, "test-delete").await?;
1132        assert!(list.is_empty());
1133
1134        Ok(())
1135    }
1136
1137    #[tokio::test]
1138    async fn delete_nonexistent() -> Result<(), Box<dyn std::error::Error>> {
1139        let bucket = crate::read_test_bucket().await;
1140
1141        let nonexistent_object = "test-delete-nonexistent";
1142
1143        let delete_result = Object::delete(&bucket.name, nonexistent_object).await;
1144
1145        if let Err(Error::Google(google_error_response)) = delete_result {
1146            assert!(google_error_response.to_string().contains(&format!(
1147                "No such object: {}/{}",
1148                bucket.name, nonexistent_object
1149            )));
1150        } else {
1151            panic!("Expected a Google error, instead got {:?}", delete_result);
1152        }
1153
1154        Ok(())
1155    }
1156
1157    #[tokio::test]
1158    async fn compose() -> Result<(), Box<dyn std::error::Error>> {
1159        let bucket = crate::read_test_bucket().await;
1160        let obj1 = Object::create(&bucket.name, vec![0, 1], "test-compose-1", "text/plain").await?;
1161        let obj2 = Object::create(&bucket.name, vec![2, 3], "test-compose-2", "text/plain").await?;
1162        let compose_request = ComposeRequest {
1163            kind: "storage#composeRequest".to_string(),
1164            source_objects: vec![
1165                SourceObject {
1166                    name: obj1.name.clone(),
1167                    generation: None,
1168                    object_preconditions: None,
1169                },
1170                SourceObject {
1171                    name: obj2.name.clone(),
1172                    generation: None,
1173                    object_preconditions: None,
1174                },
1175            ],
1176            destination: None,
1177        };
1178        let obj3 = Object::compose(&bucket.name, &compose_request, "test-concatted-file").await?;
1179        let url = obj3.download_url(100)?;
1180        let content = reqwest::get(&url).await?.text().await?;
1181        assert_eq!(content.as_bytes(), &[0, 1, 2, 3]);
1182        Ok(())
1183    }
1184
1185    #[tokio::test]
1186    async fn copy() -> Result<(), Box<dyn std::error::Error>> {
1187        let bucket = crate::read_test_bucket().await;
1188        let original = Object::create(&bucket.name, vec![2, 3], "test-copy", "text/plain").await?;
1189        original.copy(&bucket.name, "test-copy - copy").await?;
1190        Ok(())
1191    }
1192
1193    #[tokio::test]
1194    async fn rewrite() -> Result<(), Box<dyn std::error::Error>> {
1195        let bucket = crate::read_test_bucket().await;
1196        let obj = Object::create(&bucket.name, vec![0, 1], "test-rewrite", "text/plain").await?;
1197        let obj = obj.rewrite(&bucket.name, "test-rewritten").await?;
1198        let url = obj.download_url(100)?;
1199        let client = reqwest::Client::default();
1200        let download = client.head(&url).send().await?;
1201        assert_eq!(download.status().as_u16(), 200);
1202        Ok(())
1203    }
1204
1205    #[tokio::test]
1206    async fn test_url_encoding() -> Result<(), Box<dyn std::error::Error>> {
1207        let bucket = crate::read_test_bucket().await;
1208        let complicated_names = [
1209            "asdf",
1210            "asdf+1",
1211            "asdf&&+1?=3,,-_()*&^%$#@!`~{}[]\\|:;\"'<>,.?/äöüëß",
1212            "https://www.google.com",
1213            "परिक्षण फाईल",
1214            "测试很重要",
1215        ];
1216        for name in &complicated_names {
1217            let _obj = Object::create(&bucket.name, vec![0, 1], name, "text/plain").await?;
1218            let obj = Object::read(&bucket.name, &name).await.unwrap();
1219            let url = obj.download_url(100)?;
1220            let client = reqwest::Client::default();
1221            let download = client.head(&url).send().await?;
1222            assert_eq!(download.status().as_u16(), 200);
1223        }
1224        Ok(())
1225    }
1226
1227    #[tokio::test]
1228    async fn test_download_url_with() -> Result<(), Box<dyn std::error::Error>> {
1229        let bucket = crate::read_test_bucket().await;
1230        let client = reqwest::Client::new();
1231        let obj = Object::create(&bucket.name, vec![0, 1], "test-rewrite", "text/plain").await?;
1232
1233        let opts1 = crate::DownloadOptions::new().content_disposition("attachment");
1234        let download_url1 = obj.download_url_with(100, opts1)?;
1235        let download1 = client.head(&download_url1).send().await?;
1236        assert_eq!(download1.headers()["content-disposition"], "attachment");
1237        Ok(())
1238    }
1239
1240    #[tokio::test]
1241    async fn test_upload_url() -> Result<(), Box<dyn std::error::Error>> {
1242        let bucket = crate::read_test_bucket().await;
1243        let client = reqwest::Client::new();
1244        let blob_name = "test-upload-url";
1245        let obj = Object::create(&bucket.name, vec![0, 1], blob_name, "text/plain").await?;
1246
1247        let url = obj.upload_url(100).unwrap();
1248        let updated_content = vec![2, 3];
1249        let response = client
1250            .put(&url)
1251            .body(updated_content.clone())
1252            .send()
1253            .await?;
1254        assert!(response.status().is_success());
1255        let data = Object::download(&bucket.name, blob_name).await?;
1256        assert_eq!(data, updated_content);
1257        Ok(())
1258    }
1259
1260    #[tokio::test]
1261    async fn test_upload_url_with() -> Result<(), Box<dyn std::error::Error>> {
1262        let bucket = crate::read_test_bucket().await;
1263        let client = reqwest::Client::new();
1264        let blob_name = "test-upload-url";
1265        let obj = Object::create(&bucket.name, vec![0, 1], blob_name, "text/plain").await?;
1266        let mut custom_metadata = HashMap::new();
1267        custom_metadata.insert(String::from("field"), String::from("value"));
1268
1269        let (url, headers) = obj.upload_url_with(100, custom_metadata).unwrap();
1270        let updated_content = vec![2, 3];
1271        let mut request = client.put(&url).body(updated_content);
1272        for (metadata_field, metadata_value) in headers.iter() {
1273            request = request.header(metadata_field, metadata_value);
1274        }
1275        let response = request.send().await?;
1276        assert!(response.status().is_success());
1277        let updated_obj = Object::read(&bucket.name, blob_name).await?;
1278        let obj_metadata = updated_obj.metadata.unwrap();
1279        assert_eq!(obj_metadata.get("field").unwrap(), "value");
1280        Ok(())
1281    }
1282
1283    #[cfg(all(feature = "openssl", feature = "ring"))]
1284    #[test]
1285    fn check_matching_crypto() {
1286        assert_eq!(
1287            openssl::sha256(b"hello").as_ref(),
1288            ring::sha256(b"hello").as_ref()
1289        );
1290
1291        assert_eq!(
1292            openssl::rsa_pkcs1_sha256("world").unwrap(),
1293            ring::rsa_pkcs1_sha256("world").unwrap(),
1294        );
1295    }
1296
1297    #[cfg(feature = "sync")]
1298    mod sync {
1299        use super::*;
1300
1301        #[test]
1302        fn create() -> Result<(), Box<dyn std::error::Error>> {
1303            let bucket = crate::read_test_bucket_sync();
1304            Object::create_sync(&bucket.name, vec![0, 1], "test-create", "text/plain")?;
1305            Ok(())
1306        }
1307
1308        #[test]
1309        fn create_streamed() -> Result<(), Box<dyn std::error::Error>> {
1310            let bucket = crate::read_test_bucket_sync();
1311            let cursor = std::io::Cursor::new([0, 1]);
1312            Object::create_streamed_sync(
1313                &bucket.name,
1314                cursor,
1315                2,
1316                "test-create-streamed",
1317                "text/plain",
1318            )?;
1319            Ok(())
1320        }
1321
1322        #[test]
1323        fn list() -> Result<(), Box<dyn std::error::Error>> {
1324            let test_bucket = crate::read_test_bucket_sync();
1325            Object::list_sync(&test_bucket.name, ListRequest::default())?;
1326            Ok(())
1327        }
1328
1329        #[test]
1330        fn list_prefix() -> Result<(), Box<dyn std::error::Error>> {
1331            let test_bucket = crate::read_test_bucket_sync();
1332
1333            let prefix_names = [
1334                "test-list-prefix/1",
1335                "test-list-prefix/2",
1336                "test-list-prefix/sub/1",
1337                "test-list-prefix/sub/2",
1338            ];
1339
1340            for name in &prefix_names {
1341                Object::create_sync(&test_bucket.name, vec![0, 1], name, "text/plain")?;
1342            }
1343
1344            let request = ListRequest {
1345                prefix: Some("test-list-prefix/".into()),
1346                ..Default::default()
1347            };
1348            let list = Object::list_sync(&test_bucket.name, request)?;
1349            assert_eq!(list[0].items.len(), 4);
1350
1351            let request = ListRequest {
1352                prefix: Some("test-list-prefix/sub".into()),
1353                ..Default::default()
1354            };
1355            let list = Object::list_sync(&test_bucket.name, request)?;
1356            assert_eq!(list[0].items.len(), 2);
1357            Ok(())
1358        }
1359
1360        #[test]
1361        fn list_prefix_delimiter() -> Result<(), Box<dyn std::error::Error>> {
1362            let test_bucket = crate::read_test_bucket_sync();
1363
1364            let prefix_names = [
1365                "test-list-prefix/1",
1366                "test-list-prefix/2",
1367                "test-list-prefix/sub/1",
1368                "test-list-prefix/sub/2",
1369            ];
1370
1371            for name in &prefix_names {
1372                Object::create_sync(&test_bucket.name, vec![0, 1], name, "text/plain")?;
1373            }
1374
1375            let request = ListRequest {
1376                prefix: Some("test-list-prefix/".into()),
1377                delimiter: Some("/".into()),
1378                ..Default::default()
1379            };
1380            let list = Object::list_sync(&test_bucket.name, request)?;
1381            assert_eq!(list[0].items.len(), 2);
1382            assert_eq!(list[0].prefixes.len(), 1);
1383            Ok(())
1384        }
1385
1386        #[test]
1387        fn read() -> Result<(), Box<dyn std::error::Error>> {
1388            let bucket = crate::read_test_bucket_sync();
1389            Object::create_sync(&bucket.name, vec![0, 1], "test-read", "text/plain")?;
1390            Object::read_sync(&bucket.name, "test-read")?;
1391            Ok(())
1392        }
1393
1394        #[test]
1395        fn download() -> Result<(), Box<dyn std::error::Error>> {
1396            let bucket = crate::read_test_bucket_sync();
1397            let content = b"hello world";
1398            Object::create_sync(
1399                &bucket.name,
1400                content.to_vec(),
1401                "test-download",
1402                "application/octet-stream",
1403            )?;
1404
1405            let data = Object::download_sync(&bucket.name, "test-download")?;
1406            assert_eq!(data, content);
1407
1408            Ok(())
1409        }
1410
1411        #[test]
1412        fn update() -> Result<(), Box<dyn std::error::Error>> {
1413            let bucket = crate::read_test_bucket_sync();
1414            let mut obj =
1415                Object::create_sync(&bucket.name, vec![0, 1], "test-update", "text/plain")?;
1416            obj.content_type = Some("application/xml".to_string());
1417            obj.update_sync()?;
1418            Ok(())
1419        }
1420
1421        #[test]
1422        fn delete() -> Result<(), Box<dyn std::error::Error>> {
1423            let bucket = crate::read_test_bucket_sync();
1424            Object::create_sync(&bucket.name, vec![0, 1], "test-delete", "text/plain")?;
1425
1426            Object::delete_sync(&bucket.name, "test-delete")?;
1427
1428            let request = ListRequest {
1429                prefix: Some("test-delete".into()),
1430                ..Default::default()
1431            };
1432
1433            let list = Object::list_sync(&bucket.name, request)?;
1434            assert!(list[0].items.is_empty());
1435
1436            Ok(())
1437        }
1438
1439        #[test]
1440        fn delete_nonexistent() -> Result<(), Box<dyn std::error::Error>> {
1441            let bucket = crate::read_test_bucket_sync();
1442
1443            let nonexistent_object = "test-delete-nonexistent";
1444
1445            let delete_result = Object::delete_sync(&bucket.name, nonexistent_object);
1446
1447            if let Err(Error::Google(google_error_response)) = delete_result {
1448                assert!(google_error_response.to_string().contains(&format!(
1449                    "No such object: {}/{}",
1450                    bucket.name, nonexistent_object
1451                )));
1452            } else {
1453                panic!("Expected a Google error, instead got {:?}", delete_result);
1454            }
1455
1456            Ok(())
1457        }
1458
1459        #[test]
1460        fn compose() -> Result<(), Box<dyn std::error::Error>> {
1461            let bucket = crate::read_test_bucket_sync();
1462            let obj1 =
1463                Object::create_sync(&bucket.name, vec![0, 1], "test-compose-1", "text/plain")?;
1464            let obj2 =
1465                Object::create_sync(&bucket.name, vec![2, 3], "test-compose-2", "text/plain")?;
1466            let compose_request = ComposeRequest {
1467                kind: "storage#composeRequest".to_string(),
1468                source_objects: vec![
1469                    SourceObject {
1470                        name: obj1.name.clone(),
1471                        generation: None,
1472                        object_preconditions: None,
1473                    },
1474                    SourceObject {
1475                        name: obj2.name.clone(),
1476                        generation: None,
1477                        object_preconditions: None,
1478                    },
1479                ],
1480                destination: None,
1481            };
1482            let obj3 = Object::compose_sync(&bucket.name, &compose_request, "test-concatted-file")?;
1483            let url = obj3.download_url(100)?;
1484            let content = reqwest::blocking::get(&url)?.text()?;
1485            assert_eq!(content.as_bytes(), &[0, 1, 2, 3]);
1486            Ok(())
1487        }
1488
1489        #[test]
1490        fn copy() -> Result<(), Box<dyn std::error::Error>> {
1491            let bucket = crate::read_test_bucket_sync();
1492            let original =
1493                Object::create_sync(&bucket.name, vec![2, 3], "test-copy", "text/plain")?;
1494            original.copy_sync(&bucket.name, "test-copy - copy")?;
1495            Ok(())
1496        }
1497
1498        #[test]
1499        fn rewrite() -> Result<(), Box<dyn std::error::Error>> {
1500            let bucket = crate::read_test_bucket_sync();
1501            let obj = Object::create_sync(&bucket.name, vec![0, 1], "test-rewrite", "text/plain")?;
1502            let obj = obj.rewrite_sync(&bucket.name, "test-rewritten")?;
1503            let url = obj.download_url(100)?;
1504            let client = reqwest::blocking::Client::new();
1505            let download = client.head(&url).send()?;
1506            assert_eq!(download.status().as_u16(), 200);
1507            Ok(())
1508        }
1509
1510        #[test]
1511        fn test_url_encoding() -> Result<(), Box<dyn std::error::Error>> {
1512            let bucket = crate::read_test_bucket_sync();
1513            let complicated_names = [
1514                "asdf",
1515                "asdf+1",
1516                "asdf&&+1?=3,,-_()*&^%$#@!`~{}[]\\|:;\"'<>,.?/äöüëß",
1517                "https://www.google.com",
1518                "परिक्षण फाईल",
1519                "测试很重要",
1520            ];
1521            for name in &complicated_names {
1522                let _obj = Object::create_sync(&bucket.name, vec![0, 1], name, "text/plain")?;
1523                let obj = Object::read_sync(&bucket.name, &name).unwrap();
1524                let url = obj.download_url(100)?;
1525                let client = reqwest::blocking::Client::new();
1526                let download = client.head(&url).send()?;
1527                assert_eq!(download.status().as_u16(), 200);
1528            }
1529            Ok(())
1530        }
1531    }
1532}
1533
1534/// A wrapper around a downloaded object's byte stream that provides a useful `size_hint`.
1535pub struct SizedByteStream<S: Stream<Item = crate::Result<u8>> + Unpin> {
1536    size: Option<u64>,
1537    bytes: S,
1538}
1539
1540impl<S: Stream<Item = crate::Result<u8>> + Unpin> SizedByteStream<S> {
1541    pub(crate) fn new(bytes: S, size: Option<u64>) -> Self {
1542        Self { size, bytes }
1543    }
1544}
1545
1546impl<S: Stream<Item = crate::Result<u8>> + Unpin> Stream for SizedByteStream<S> {
1547    type Item = crate::Result<u8>;
1548
1549    fn poll_next(
1550        mut self: std::pin::Pin<&mut Self>,
1551        cx: &mut std::task::Context,
1552    ) -> std::task::Poll<Option<Self::Item>> {
1553        futures_util::StreamExt::poll_next_unpin(&mut self.bytes, cx)
1554    }
1555
1556    fn size_hint(&self) -> (usize, Option<usize>) {
1557        let size = self
1558            .size
1559            .and_then(|s| std::convert::TryInto::try_into(s).ok());
1560        (size.unwrap_or(0), size)
1561    }
1562}
cloud_storage/resources/object.rs

cloud_storage/resources/
object.rs
