google_cloud_storage/storage/write_object.rs
1// Copyright 2025 Google LLC
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// https://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Contains the request builder for [write_object()] and related types.
16//!
17//! [write_object()]: crate::storage::client::Storage::write_object()
18
19use super::streaming_source::{Seek, StreamingSource};
20use super::*;
21use crate::model_ext::KeyAes256;
22use crate::storage::checksum::details::update as checksum_update;
23use crate::storage::checksum::details::{Checksum, Md5};
24use crate::storage::request_options::RequestOptions;
25
26/// A request builder for object writes.
27///
28/// # Example: hello world
29/// ```
30/// use google_cloud_storage::client::Storage;
31/// async fn sample(client: &Storage) -> anyhow::Result<()> {
32/// let response = client
33/// .write_object("projects/_/buckets/my-bucket", "hello", "Hello World!")
34/// .send_unbuffered()
35/// .await?;
36/// println!("response details={response:?}");
37/// Ok(())
38/// }
39/// ```
40///
41/// # Example: upload a file
42/// ```
43/// use google_cloud_storage::client::Storage;
44/// async fn sample(client: &Storage) -> anyhow::Result<()> {
45/// let payload = tokio::fs::File::open("my-data").await?;
46/// let response = client
47/// .write_object("projects/_/buckets/my-bucket", "my-object", payload)
48/// .send_unbuffered()
49/// .await?;
50/// println!("response details={response:?}");
51/// Ok(())
52/// }
53/// ```
54///
55/// # Example: create a new object from a custom data source
56/// ```
57/// use google_cloud_storage::{client::Storage, streaming_source::StreamingSource};
58/// struct DataSource;
59/// impl StreamingSource for DataSource {
60/// type Error = std::io::Error;
61/// async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
62/// # panic!();
63/// }
64/// }
65///
66/// async fn sample(client: &Storage) -> anyhow::Result<()> {
67/// let response = client
68/// .write_object("projects/_/buckets/my-bucket", "my-object", DataSource)
69/// .send_buffered()
70/// .await?;
71/// println!("response details={response:?}");
72/// Ok(())
73/// }
74/// ```
75pub struct WriteObject<T, S = crate::storage::transport::Storage>
76where
77 S: crate::storage::stub::Storage + 'static,
78{
79 stub: std::sync::Arc<S>,
80 pub(crate) request: crate::model_ext::WriteObjectRequest,
81 pub(crate) payload: Payload<T>,
82 pub(crate) options: RequestOptions,
83}
84
85impl<T, S> WriteObject<T, S>
86where
87 S: crate::storage::stub::Storage + 'static,
88{
89 /// Set a [request precondition] on the object generation to match.
90 ///
91 /// With this precondition the request fails if the current object
92 /// generation matches the provided value. A common value is `0`, which
93 /// prevents writes from succeeding if the object already exists.
94 ///
95 /// # Example
96 /// ```
97 /// # use google_cloud_storage::client::Storage;
98 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
99 /// let response = client
100 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
101 /// .set_if_generation_match(0)
102 /// .send_buffered()
103 /// .await?;
104 /// println!("response details={response:?}");
105 /// # Ok(()) }
106 /// ```
107 ///
108 /// [request precondition]: https://cloud.google.com/storage/docs/request-preconditions
109 pub fn set_if_generation_match<V>(mut self, v: V) -> Self
110 where
111 V: Into<i64>,
112 {
113 self.request.spec.if_generation_match = Some(v.into());
114 self
115 }
116
117 /// Set a [request precondition] on the object generation to match.
118 ///
119 /// With this precondition the request fails if the current object
120 /// generation does not match the provided value.
121 ///
122 /// # Example
123 /// ```
124 /// # use google_cloud_storage::client::Storage;
125 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
126 /// let response = client
127 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
128 /// .set_if_generation_not_match(0)
129 /// .send_buffered()
130 /// .await?;
131 /// println!("response details={response:?}");
132 /// # Ok(()) }
133 /// ```
134 ///
135 /// [request precondition]: https://cloud.google.com/storage/docs/request-preconditions
136 pub fn set_if_generation_not_match<V>(mut self, v: V) -> Self
137 where
138 V: Into<i64>,
139 {
140 self.request.spec.if_generation_not_match = Some(v.into());
141 self
142 }
143
144 /// Set a [request precondition] on the object meta generation.
145 ///
146 /// With this precondition the request fails if the current object metadata
147 /// generation does not match the provided value. This may be useful to
148 /// prevent changes when the metageneration is known.
149 ///
150 /// # Example
151 /// ```
152 /// # use google_cloud_storage::client::Storage;
153 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
154 /// let response = client
155 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
156 /// .set_if_metageneration_match(1234)
157 /// .send_buffered()
158 /// .await?;
159 /// println!("response details={response:?}");
160 /// # Ok(()) }
161 /// ```
162 ///
163 /// [request precondition]: https://cloud.google.com/storage/docs/request-preconditions
164 pub fn set_if_metageneration_match<V>(mut self, v: V) -> Self
165 where
166 V: Into<i64>,
167 {
168 self.request.spec.if_metageneration_match = Some(v.into());
169 self
170 }
171
172 /// Set a [request precondition] on the object meta-generation.
173 ///
174 /// With this precondition the request fails if the current object metadata
175 /// generation matches the provided value. This is rarely useful in uploads,
176 /// it is more commonly used on reads to prevent a large response if the
177 /// data is already cached.
178 ///
179 /// # Example
180 /// ```
181 /// # use google_cloud_storage::client::Storage;
182 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
183 /// let response = client
184 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
185 /// .set_if_metageneration_not_match(1234)
186 /// .send_buffered()
187 /// .await?;
188 /// println!("response details={response:?}");
189 /// # Ok(()) }
190 /// ```
191 ///
192 /// [request precondition]: https://cloud.google.com/storage/docs/request-preconditions
193 pub fn set_if_metageneration_not_match<V>(mut self, v: V) -> Self
194 where
195 V: Into<i64>,
196 {
197 self.request.spec.if_metageneration_not_match = Some(v.into());
198 self
199 }
200
201 /// Sets the ACL for the new object.
202 ///
203 /// # Example
204 /// ```
205 /// # use google_cloud_storage::client::Storage;
206 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
207 /// # use google_cloud_storage::model::ObjectAccessControl;
208 /// let response = client
209 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
210 /// .set_acl([ObjectAccessControl::new().set_entity("allAuthenticatedUsers").set_role("READER")])
211 /// .send_buffered()
212 /// .await?;
213 /// println!("response details={response:?}");
214 /// # Ok(()) }
215 /// ```
216 pub fn set_acl<I, V>(mut self, v: I) -> Self
217 where
218 I: IntoIterator<Item = V>,
219 V: Into<crate::model::ObjectAccessControl>,
220 {
221 self.mut_resource().acl = v.into_iter().map(|a| a.into()).collect();
222 self
223 }
224
225 /// Sets the [cache control] for the new object.
226 ///
227 /// This can be used to control caching in [public objects].
228 ///
229 /// # Example
230 /// ```
231 /// # use google_cloud_storage::client::Storage;
232 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
233 /// let response = client
234 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
235 /// .set_cache_control("public; max-age=7200")
236 /// .send_buffered()
237 /// .await?;
238 /// println!("response details={response:?}");
239 /// # Ok(()) }
240 /// ```
241 ///
242 /// [public objects]: https://cloud.google.com/storage/docs/access-control/making-data-public
243 /// [cache control]: https://datatracker.ietf.org/doc/html/rfc7234#section-5.2
244 pub fn set_cache_control<V: Into<String>>(mut self, v: V) -> Self {
245 self.mut_resource().cache_control = v.into();
246 self
247 }
248
249 /// Sets the [content disposition] for the new object.
250 ///
251 /// Google Cloud Storage can serve content directly to web browsers. This
252 /// attribute sets the `Content-Disposition` header, which may change how
253 /// the browser displays the contents.
254 ///
255 /// # Example
256 /// ```
257 /// # use google_cloud_storage::client::Storage;
258 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
259 /// let response = client
260 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
261 /// .set_content_disposition("inline")
262 /// .send_buffered()
263 /// .await?;
264 /// println!("response details={response:?}");
265 /// # Ok(()) }
266 /// ```
267 ///
268 /// [content disposition]: https://datatracker.ietf.org/doc/html/rfc6266
269 pub fn set_content_disposition<V: Into<String>>(mut self, v: V) -> Self {
270 self.mut_resource().content_disposition = v.into();
271 self
272 }
273
274 /// Sets the [content encoding] for the object data.
275 ///
276 /// This can be used to upload compressed data and enable [transcoding] of
277 /// the data during reads.
278 ///
279 /// # Example
280 /// ```
281 /// # use google_cloud_storage::client::Storage;
282 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
283 /// use flate2::write::GzEncoder;
284 /// use std::io::Write;
285 /// let mut e = GzEncoder::new(Vec::new(), flate2::Compression::default());
286 /// e.write_all(b"hello world");
287 /// let response = client
288 /// .write_object("projects/_/buckets/my-bucket", "my-object", bytes::Bytes::from_owner(e.finish()?))
289 /// .set_content_encoding("gzip")
290 /// .send_buffered()
291 /// .await?;
292 /// println!("response details={response:?}");
293 /// # Ok(()) }
294 /// ```
295 ///
296 /// [transcoding]: https://cloud.google.com/storage/docs/transcoding
297 /// [content encoding]: https://datatracker.ietf.org/doc/html/rfc7231#section-3.1.2.2
298 pub fn set_content_encoding<V: Into<String>>(mut self, v: V) -> Self {
299 self.mut_resource().content_encoding = v.into();
300 self
301 }
302
303 /// Sets the [content language] for the new object.
304 ///
305 /// Google Cloud Storage can serve content directly to web browsers. This
306 /// attribute sets the `Content-Language` header, which may change how the
307 /// browser displays the contents.
308 ///
309 /// # Example
310 /// ```
311 /// # use google_cloud_storage::client::Storage;
312 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
313 /// let response = client
314 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
315 /// .set_content_language("en")
316 /// .send_buffered()
317 /// .await?;
318 /// println!("response details={response:?}");
319 /// # Ok(()) }
320 /// ```
321 ///
322 /// [content language]: https://cloud.google.com/storage/docs/metadata#content-language
323 pub fn set_content_language<V: Into<String>>(mut self, v: V) -> Self {
324 self.mut_resource().content_language = v.into();
325 self
326 }
327
328 /// Sets the [content type] for the new object.
329 ///
330 /// Google Cloud Storage can serve content directly to web browsers. This
331 /// attribute sets the `Content-Type` header, which may change how the
332 /// browser interprets the contents.
333 ///
334 /// # Example
335 /// ```
336 /// # use google_cloud_storage::client::Storage;
337 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
338 /// let response = client
339 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
340 /// .set_content_type("text/plain")
341 /// .send_buffered()
342 /// .await?;
343 /// println!("response details={response:?}");
344 /// # Ok(()) }
345 /// ```
346 ///
347 /// [content type]: https://datatracker.ietf.org/doc/html/rfc7231#section-3.1.1.5
348 pub fn set_content_type<V: Into<String>>(mut self, v: V) -> Self {
349 self.mut_resource().content_type = v.into();
350 self
351 }
352
353 /// Sets the [custom time] for the new object.
354 ///
355 /// This field is typically set in order to use the [DaysSinceCustomTime]
356 /// condition in Object Lifecycle Management.
357 ///
358 /// # Example
359 /// ```
360 /// # use google_cloud_storage::client::Storage;
361 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
362 /// let time = wkt::Timestamp::try_from("2025-07-07T18:30:00Z")?;
363 /// let response = client
364 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
365 /// .set_custom_time(time)
366 /// .send_buffered()
367 /// .await?;
368 /// println!("response details={response:?}");
369 /// # Ok(()) }
370 /// ```
371 ///
372 /// [DaysSinceCustomTime]: https://cloud.google.com/storage/docs/lifecycle#dayssincecustomtime
373 /// [custom time]: https://cloud.google.com/storage/docs/metadata#custom-time
374 pub fn set_custom_time<V: Into<wkt::Timestamp>>(mut self, v: V) -> Self {
375 self.mut_resource().custom_time = Some(v.into());
376 self
377 }
378
379 /// Sets the [event based hold] flag for the new object.
380 ///
381 /// This field is typically set in order to prevent objects from being
382 /// deleted or modified.
383 ///
384 /// # Example
385 /// ```
386 /// # use google_cloud_storage::client::Storage;
387 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
388 /// let response = client
389 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
390 /// .set_event_based_hold(true)
391 /// .send_buffered()
392 /// .await?;
393 /// println!("response details={response:?}");
394 /// # Ok(()) }
395 /// ```
396 ///
397 /// [event based hold]: https://cloud.google.com/storage/docs/object-holds
398 pub fn set_event_based_hold<V: Into<bool>>(mut self, v: V) -> Self {
399 self.mut_resource().event_based_hold = Some(v.into());
400 self
401 }
402
403 /// Sets the [custom metadata] for the new object.
404 ///
405 /// This field is typically set to annotate the object with
406 /// application-specific metadata.
407 ///
408 /// # Example
409 /// ```
410 /// # use google_cloud_storage::client::Storage;
411 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
412 /// let time = wkt::Timestamp::try_from("2025-07-07T18:30:00Z")?;
413 /// let response = client
414 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
415 /// .set_metadata([("test-only", "true"), ("environment", "qa")])
416 /// .send_buffered()
417 /// .await?;
418 /// println!("response details={response:?}");
419 /// # Ok(()) }
420 /// ```
421 ///
422 /// [custom metadata]: https://cloud.google.com/storage/docs/metadata#custom-metadata
423 pub fn set_metadata<I, K, V>(mut self, i: I) -> Self
424 where
425 I: IntoIterator<Item = (K, V)>,
426 K: Into<String>,
427 V: Into<String>,
428 {
429 self.mut_resource().metadata = i.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
430 self
431 }
432
433 /// Sets the [retention configuration] for the new object.
434 ///
435 /// # Example
436 /// ```
437 /// # use google_cloud_storage::client::Storage;
438 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
439 /// # use google_cloud_storage::model::object::{Retention, retention};
440 /// let response = client
441 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
442 /// .set_retention(
443 /// Retention::new()
444 /// .set_mode(retention::Mode::Locked)
445 /// .set_retain_until_time(wkt::Timestamp::try_from("2035-01-01T00:00:00Z")?))
446 /// .send_buffered()
447 /// .await?;
448 /// println!("response details={response:?}");
449 /// # Ok(()) }
450 /// ```
451 ///
452 /// [retention configuration]: https://cloud.google.com/storage/docs/metadata#retention-config
453 pub fn set_retention<V>(mut self, v: V) -> Self
454 where
455 V: Into<crate::model::object::Retention>,
456 {
457 self.mut_resource().retention = Some(v.into());
458 self
459 }
460
461 /// Sets the [storage class] for the new object.
462 ///
463 /// # Example
464 /// ```
465 /// # use google_cloud_storage::client::Storage;
466 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
467 /// let response = client
468 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
469 /// .set_storage_class("ARCHIVE")
470 /// .send_buffered()
471 /// .await?;
472 /// println!("response details={response:?}");
473 /// # Ok(()) }
474 /// ```
475 ///
476 /// [storage class]: https://cloud.google.com/storage/docs/storage-classes
477 pub fn set_storage_class<V>(mut self, v: V) -> Self
478 where
479 V: Into<String>,
480 {
481 self.mut_resource().storage_class = v.into();
482 self
483 }
484
485 /// Sets the [temporary hold] flag for the new object.
486 ///
487 /// This field is typically set in order to prevent objects from being
488 /// deleted or modified.
489 ///
490 /// # Example
491 /// ```
492 /// # use google_cloud_storage::client::Storage;
493 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
494 /// let time = wkt::Timestamp::try_from("2025-07-07T18:30:00Z")?;
495 /// let response = client
496 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
497 /// .set_temporary_hold(true)
498 /// .send_buffered()
499 /// .await?;
500 /// println!("response details={response:?}");
501 /// # Ok(()) }
502 /// ```
503 ///
504 /// [temporary hold]: https://cloud.google.com/storage/docs/object-holds
505 pub fn set_temporary_hold<V: Into<bool>>(mut self, v: V) -> Self {
506 self.mut_resource().temporary_hold = v.into();
507 self
508 }
509
510 /// Sets the resource name of the [Customer-managed encryption key] for this
511 /// object.
512 ///
513 /// The service imposes a number of restrictions on the keys used to encrypt
514 /// Google Cloud Storage objects. Read the documentation in full before
515 /// trying to use customer-managed encryption keys. In particular, verify
516 /// the service has the necessary permissions, and the key is in a
517 /// compatible location.
518 ///
519 /// # Example
520 /// ```
521 /// # use google_cloud_storage::client::Storage;
522 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
523 /// let response = client
524 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
525 /// .set_kms_key("projects/test-project/locations/us-central1/keyRings/test-ring/cryptoKeys/test-key")
526 /// .send_buffered()
527 /// .await?;
528 /// println!("response details={response:?}");
529 /// # Ok(()) }
530 /// ```
531 ///
532 /// [Customer-managed encryption key]: https://cloud.google.com/storage/docs/encryption/customer-managed-keys
533 pub fn set_kms_key<V>(mut self, v: V) -> Self
534 where
535 V: Into<String>,
536 {
537 self.mut_resource().kms_key = v.into();
538 self
539 }
540
541 /// Configure this object to use one of the [predefined ACLs].
542 ///
543 /// # Example
544 /// ```
545 /// # use google_cloud_storage::client::Storage;
546 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
547 /// let response = client
548 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
549 /// .set_predefined_acl("private")
550 /// .send_buffered()
551 /// .await?;
552 /// println!("response details={response:?}");
553 /// # Ok(()) }
554 /// ```
555 ///
556 /// [predefined ACLs]: https://cloud.google.com/storage/docs/access-control/lists#predefined-acl
557 pub fn set_predefined_acl<V>(mut self, v: V) -> Self
558 where
559 V: Into<String>,
560 {
561 self.request.spec.predefined_acl = v.into();
562 self
563 }
564
565 /// The encryption key used with the Customer-Supplied Encryption Keys
566 /// feature. In raw bytes format (not base64-encoded).
567 ///
568 /// # Example
569 /// ```
570 /// # use google_cloud_storage::client::Storage;
571 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
572 /// # use google_cloud_storage::model_ext::KeyAes256;
573 /// let key: &[u8] = &[97; 32];
574 /// let response = client
575 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
576 /// .set_key(KeyAes256::new(key)?)
577 /// .send_buffered()
578 /// .await?;
579 /// println!("response details={response:?}");
580 /// # Ok(()) }
581 /// ```
582 pub fn set_key(mut self, v: KeyAes256) -> Self {
583 self.request.params = Some(v.into());
584 self
585 }
586
587 /// Sets the object custom contexts.
588 ///
589 /// # Example
590 /// ```
591 /// # use google_cloud_storage::client::Storage;
592 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
593 /// # use google_cloud_storage::model::{ObjectContexts, ObjectCustomContextPayload};
594 /// let response = client
595 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
596 /// .set_contexts(
597 /// ObjectContexts::new().set_custom([
598 /// ("example", ObjectCustomContextPayload::new().set_value("true")),
599 /// ])
600 /// )
601 /// .send_buffered()
602 /// .await?;
603 /// println!("response details={response:?}");
604 /// # Ok(()) }
605 /// ```
606 pub fn set_contexts<V>(mut self, v: V) -> Self
607 where
608 V: Into<crate::model::ObjectContexts>,
609 {
610 self.mut_resource().contexts = Some(v.into());
611 self
612 }
613
614 /// Configure the idempotency for this upload.
615 ///
616 /// By default, the client library treats single-shot uploads without an
617 /// `if_generation_match` precondition as non-idempotent. If the destination
618 /// bucket is configured with [object versioning] then the operation may
619 /// succeed multiple times with observable side-effects. With object
620 /// versioning and a [lifecycle] policy limiting the number of versions,
621 /// uploading the same data multiple times may result in data loss.
622 ///
623 /// The client library cannot efficiently determine if these conditions
624 /// apply to your upload. If they do not, or your application can tolerate
625 /// multiple versions of the same data for other reasons, consider using
626 /// `with_idempotency(true)`. Use `with_idempotency(false)` to disable
627 /// retries even when `if_generation_match` is set.
628 ///
629 /// The client library treats resumable uploads as idempotent, regardless of
630 /// the value in this option. Such uploads can succeed at most once.
631 ///
632 /// # Example
633 /// ```
634 /// # use google_cloud_storage::client::Storage;
635 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
636 /// use std::time::Duration;
637 /// use google_cloud_gax::retry_policy::RetryPolicyExt;
638 /// let response = client
639 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
640 /// .with_idempotency(true)
641 /// .send_buffered()
642 /// .await?;
643 /// println!("response details={response:?}");
644 /// # Ok(()) }
645 /// ```
646 ///
647 /// [lifecycle]: https://cloud.google.com/storage/docs/lifecycle
648 /// [object versioning]: https://cloud.google.com/storage/docs/object-versioning
649 pub fn with_idempotency(mut self, v: bool) -> Self {
650 self.options.idempotency = Some(v);
651 self
652 }
653
654 /// The retry policy used for this request.
655 ///
656 /// # Example
657 /// ```
658 /// # use google_cloud_storage::client::Storage;
659 /// # use google_cloud_storage::retry_policy::RetryableErrors;
660 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
661 /// use std::time::Duration;
662 /// use google_cloud_gax::retry_policy::RetryPolicyExt;
663 /// let response = client
664 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
665 /// .with_retry_policy(
666 /// RetryableErrors
667 /// .with_attempt_limit(5)
668 /// .with_time_limit(Duration::from_secs(90)),
669 /// )
670 /// .send_buffered()
671 /// .await?;
672 /// println!("response details={response:?}");
673 /// # Ok(()) }
674 /// ```
675 pub fn with_retry_policy<V: Into<google_cloud_gax::retry_policy::RetryPolicyArg>>(
676 mut self,
677 v: V,
678 ) -> Self {
679 self.options.retry_policy = v.into().into();
680 self
681 }
682
683 /// The backoff policy used for this request.
684 ///
685 /// # Example
686 /// ```
687 /// # use google_cloud_storage::client::Storage;
688 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
689 /// use std::time::Duration;
690 /// use google_cloud_gax::exponential_backoff::ExponentialBackoff;
691 /// let response = client
692 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
693 /// .with_backoff_policy(ExponentialBackoff::default())
694 /// .send_buffered()
695 /// .await?;
696 /// println!("response details={response:?}");
697 /// # Ok(()) }
698 /// ```
699 pub fn with_backoff_policy<V: Into<google_cloud_gax::backoff_policy::BackoffPolicyArg>>(
700 mut self,
701 v: V,
702 ) -> Self {
703 self.options.backoff_policy = v.into().into();
704 self
705 }
706
707 /// The retry throttler used for this request.
708 ///
709 /// Most of the time you want to use the same throttler for all the requests
710 /// in a client, and even the same throttler for many clients. Rarely it
711 /// may be necessary to use an custom throttler for some subset of the
712 /// requests.
713 ///
714 /// # Example
715 /// ```
716 /// # use google_cloud_storage::client::Storage;
717 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
718 /// let response = client
719 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
720 /// .with_retry_throttler(adhoc_throttler())
721 /// .send_buffered()
722 /// .await?;
723 /// println!("response details={response:?}");
724 /// fn adhoc_throttler() -> google_cloud_gax::retry_throttler::SharedRetryThrottler {
725 /// # panic!();
726 /// }
727 /// # Ok(()) }
728 /// ```
729 pub fn with_retry_throttler<V: Into<google_cloud_gax::retry_throttler::RetryThrottlerArg>>(
730 mut self,
731 v: V,
732 ) -> Self {
733 self.options.retry_throttler = v.into().into();
734 self
735 }
736
737 /// Sets the payload size threshold to switch from single-shot to resumable uploads.
738 ///
739 /// # Example
740 /// ```
741 /// # use google_cloud_storage::client::Storage;
742 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
743 /// let response = client
744 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
745 /// .with_resumable_upload_threshold(0_usize) // Forces a resumable upload.
746 /// .send_buffered()
747 /// .await?;
748 /// println!("response details={response:?}");
749 /// # Ok(()) }
750 /// ```
751 ///
752 /// The client library can perform uploads using [single-shot] or
753 /// [resumable] uploads. For small objects, single-shot uploads offer better
754 /// performance, as they require a single HTTP transfer. For larger objects,
755 /// the additional request latency is not significant, and resumable uploads
756 /// offer better recovery on errors.
757 ///
758 /// The library automatically selects resumable uploads when the payload is
759 /// equal to or larger than this option. For smaller uploads the client
760 /// library uses single-shot uploads.
761 ///
762 /// The exact threshold depends on where the application is deployed and
763 /// destination bucket location with respect to where the application is
764 /// running. The library defaults should work well in most cases, but some
765 /// applications may benefit from fine-tuning.
766 ///
767 /// [single-shot]: https://cloud.google.com/storage/docs/uploading-objects
768 /// [resumable]: https://cloud.google.com/storage/docs/resumable-uploads
769 pub fn with_resumable_upload_threshold<V: Into<usize>>(mut self, v: V) -> Self {
770 self.options.set_resumable_upload_threshold(v.into());
771 self
772 }
773
774 /// Changes the buffer size for some resumable uploads.
775 ///
776 /// # Example
777 /// ```
778 /// # use google_cloud_storage::client::Storage;
779 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
780 /// let response = client
781 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
782 /// .with_resumable_upload_buffer_size(32 * 1024 * 1024_usize)
783 /// .send_buffered()
784 /// .await?;
785 /// println!("response details={response:?}");
786 /// # Ok(()) }
787 /// ```
788 ///
789 /// When performing [resumable uploads] from sources without [Seek] the
790 /// client library needs to buffer data in memory until it is persisted by
791 /// the service. Otherwise the data would be lost if the upload fails.
792 /// Applications may want to tune this buffer size:
793 ///
794 /// - Use smaller buffer sizes to support more concurrent uploads in the
795 /// same application.
796 /// - Use larger buffer sizes for better throughput. Sending many small
797 /// buffers stalls the upload until the client receives a successful
798 /// response from the service.
799 ///
800 /// Keep in mind that there are diminishing returns on using larger buffers.
801 ///
802 /// [resumable uploads]: https://cloud.google.com/storage/docs/resumable-uploads
803 /// [Seek]: crate::streaming_source::Seek
804 pub fn with_resumable_upload_buffer_size<V: Into<usize>>(mut self, v: V) -> Self {
805 self.options.set_resumable_upload_buffer_size(v.into());
806 self
807 }
808
809 /// Sets the `User-Agent` header for this request.
810 ///
811 /// # Example
812 /// ```
813 /// # use google_cloud_storage::client::Storage;
814 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
815 /// let mut response = client
816 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
817 /// .with_user_agent("my-app/1.0.0")
818 /// .send_buffered()
819 /// .await?;
820 /// println!("response details={response:?}");
821 /// # Ok(()) }
822 /// ```
823 pub fn with_user_agent(mut self, user_agent: impl Into<String>) -> Self {
824 self.options.user_agent = Some(user_agent.into());
825 self
826 }
827
828 /// Sets the project that will be billed for this request.
829 ///
830 /// Required for [Requester Pays] buckets. The value overrides any
831 /// `quota_project_id` configured on the credentials; the credential-level
832 /// header is suppressed for this RPC.
833 ///
834 /// # Example
835 /// ```
836 /// # use google_cloud_storage::client::Storage;
837 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
838 /// let response = client
839 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello")
840 /// .with_quota_project("my-billing-project")
841 /// .send_buffered()
842 /// .await?;
843 /// # Ok(()) }
844 /// ```
845 ///
846 /// [Requester Pays]: https://cloud.google.com/storage/docs/requester-pays
847 pub fn with_quota_project(mut self, project: impl Into<String>) -> Self {
848 self.options.set_quota_project(project);
849 self
850 }
851
852 fn mut_resource(&mut self) -> &mut crate::model::Object {
853 self.request
854 .spec
855 .resource
856 .as_mut()
857 .expect("resource field initialized in `new()`")
858 }
859
860 fn set_crc32c<V: Into<u32>>(mut self, v: V) -> Self {
861 let checksum = self.mut_resource().checksums.get_or_insert_default();
862 checksum.crc32c = Some(v.into());
863 self
864 }
865
866 /// Sets the MD5 hash for the object being written.
867 pub fn set_md5_hash<I, V>(mut self, i: I) -> Self
868 where
869 I: IntoIterator<Item = V>,
870 V: Into<u8>,
871 {
872 let checksum = self.mut_resource().checksums.get_or_insert_default();
873 checksum.md5_hash = i.into_iter().map(|v| v.into()).collect();
874 self
875 }
876
877 /// Provide a precomputed value for the CRC32C checksum.
878 ///
879 /// # Example
880 /// ```
881 /// # use google_cloud_storage::client::Storage;
882 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
883 /// use crc32c::crc32c;
884 /// let response = client
885 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
886 /// .with_known_crc32c(crc32c(b"hello world"))
887 /// .send_buffered()
888 /// .await?;
889 /// println!("response details={response:?}");
890 /// # Ok(()) }
891 /// ```
892 ///
893 /// In some applications, the payload's CRC32C checksum is already known.
894 /// For example, the application may be reading the data from another blob
895 /// storage system.
896 ///
897 /// In such cases, it is safer to pass the known CRC32C of the payload to
898 /// [Cloud Storage], and more efficient to skip the computation in the
899 /// client library.
900 ///
901 /// Note that once you provide a CRC32C value to this builder you cannot
902 /// use [compute_md5()] to also have the library compute the checksums.
903 ///
904 /// [compute_md5()]: WriteObject::compute_md5
905 pub fn with_known_crc32c<V: Into<u32>>(self, v: V) -> Self {
906 let mut this = self;
907 this.options.checksum.crc32c = None;
908 this.set_crc32c(v)
909 }
910
911 /// Provide a precomputed value for the MD5 hash.
912 ///
913 /// # Example
914 /// ```
915 /// # use google_cloud_storage::client::Storage;
916 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
917 /// use md5::compute;
918 /// let hash = md5::compute(b"hello world");
919 /// let response = client
920 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
921 /// .with_known_md5_hash(bytes::Bytes::from_owner(hash.0))
922 /// .send_buffered()
923 /// .await?;
924 /// println!("response details={response:?}");
925 /// # Ok(()) }
926 /// ```
927 ///
928 /// In some applications, the payload's MD5 hash is already known. For
929 /// example, the application may be reading the data from another blob
930 /// storage system.
931 ///
932 /// In such cases, it is safer to pass the known MD5 of the payload to
933 /// [Cloud Storage], and more efficient to skip the computation in the
934 /// client library.
935 ///
936 /// Note that once you provide a MD5 value to this builder you cannot
937 /// use [compute_md5()] to also have the library compute the checksums.
938 ///
939 /// [compute_md5()]: WriteObject::compute_md5
940 pub fn with_known_md5_hash<I, V>(self, i: I) -> Self
941 where
942 I: IntoIterator<Item = V>,
943 V: Into<u8>,
944 {
945 let mut this = self;
946 this.options.checksum.md5_hash = None;
947 this.set_md5_hash(i)
948 }
949
950 /// Enables computation of MD5 hashes.
951 ///
952 /// # Example
953 /// ```
954 /// # use google_cloud_storage::client::Storage;
955 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
956 /// let payload = tokio::fs::File::open("my-data").await?;
957 /// let response = client
958 /// .write_object("projects/_/buckets/my-bucket", "my-object", payload)
959 /// .compute_md5()
960 /// .send_buffered()
961 /// .await?;
962 /// println!("response details={response:?}");
963 /// # Ok(()) }
964 /// ```
965 ///
966 /// See [precompute_checksums][WriteObject::precompute_checksums] for more
967 /// details on how checksums are used by the client library and their
968 /// limitations.
969 pub fn compute_md5(self) -> Self {
970 let mut this = self;
971 this.options.checksum.md5_hash = Some(Md5::default());
972 this
973 }
974
975 pub(crate) fn new<B, O, P>(
976 stub: std::sync::Arc<S>,
977 bucket: B,
978 object: O,
979 payload: P,
980 options: RequestOptions,
981 ) -> Self
982 where
983 B: Into<String>,
984 O: Into<String>,
985 P: Into<Payload<T>>,
986 {
987 let resource = crate::model::Object::new()
988 .set_bucket(bucket)
989 .set_name(object);
990 WriteObject {
991 stub,
992 request: crate::model_ext::WriteObjectRequest {
993 spec: crate::model::WriteObjectSpec::new().set_resource(resource),
994 params: None,
995 checksum_precomputation: true,
996 },
997 payload: payload.into(),
998 options,
999 }
1000 }
1001}
1002
1003impl<T, S> WriteObject<T, S>
1004where
1005 T: StreamingSource + Seek + Send + Sync + 'static,
1006 <T as StreamingSource>::Error: std::error::Error + Send + Sync + 'static,
1007 <T as Seek>::Error: std::error::Error + Send + Sync + 'static,
1008 S: crate::storage::stub::Storage + 'static,
1009{
1010 /// A simple upload from a buffer.
1011 ///
1012 /// # Example
1013 /// ```
1014 /// # use google_cloud_storage::client::Storage;
1015 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
1016 /// let response = client
1017 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
1018 /// .send_unbuffered()
1019 /// .await?;
1020 /// println!("response details={response:?}");
1021 /// # Ok(()) }
1022 /// ```
1023 pub async fn send_unbuffered(self) -> Result<Object> {
1024 self.stub
1025 .write_object_unbuffered(self.payload, self.request, self.options)
1026 .await
1027 }
1028
1029 /// Configures checksum precomputation for unbuffered resumable uploads.
1030 ///
1031 /// Checksum precomputation is **enabled by default (`true`)** to ensure server-side data
1032 /// integrity validation for large objects streamed via resumable uploads.
1033 ///
1034 /// Note: Single-shot unbuffered uploads (< resumable upload threshold) always compute checksums
1035 /// on the fly via trailing multipart metadata and do not require precomputation.
1036 ///
1037 /// Call `.with_checksum_precomputation(false)` to turn off precomputation and prioritize
1038 /// upload speed.
1039 ///
1040 /// # Example
1041 /// ```
1042 /// # use google_cloud_storage::client::Storage;
1043 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
1044 /// let payload = tokio::fs::File::open("my-data").await?;
1045 /// let response = client
1046 /// .write_object("projects/_/buckets/my-bucket", "my-object", payload)
1047 /// .with_checksum_precomputation(false)
1048 /// .send_unbuffered()
1049 /// .await?;
1050 /// println!("response details={response:?}");
1051 /// # Ok(()) }
1052 /// ```
1053 pub fn with_checksum_precomputation(mut self, enable: bool) -> Self {
1054 self.request.checksum_precomputation = enable;
1055 self
1056 }
1057
1058 /// Precomputes the payload checksums before uploading the data.
1059 ///
1060 /// If the checksums are known when the upload starts, the client library
1061 /// includes them with the upload request, allowing the service to reject
1062 /// the upload if the payload and the checksums do not match.
1063 ///
1064 /// # Example
1065 /// ```
1066 /// # use google_cloud_storage::client::Storage;
1067 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
1068 /// let payload = tokio::fs::File::open("my-data").await?;
1069 /// #[allow(deprecated)]
1070 /// let response = client
1071 /// .write_object("projects/_/buckets/my-bucket", "my-object", payload)
1072 /// .precompute_checksums()
1073 /// .await?
1074 /// .send_unbuffered()
1075 /// .await?;
1076 /// println!("response details={response:?}");
1077 /// # Ok(()) }
1078 /// ```
1079 ///
1080 /// # Deprecated
1081 /// `precompute_checksums()` is now redundant because checksums are automatically
1082 /// validated by default across all upload modes:
1083 /// - **Buffered upload**: Checksum is uploaded in the final chunk.
1084 /// - **Single-shot unbuffered upload**: Checksum is attached as trailing metadata in a
1085 /// multipart request.
1086 /// - **Resumable unbuffered upload**: Checksum is precomputed upfront by default
1087 /// (or supplied via `with_known_crc32c`).
1088 ///
1089 /// For unbuffered resumable uploads, precomputing checksums incurs an extra read pass.
1090 /// If you want to prioritize upload speed over data integrity, call
1091 /// [with_checksum_precomputation(false)][WriteObject::with_checksum_precomputation].
1092 #[deprecated(
1093 since = "1.19.0",
1094 note = "`precompute_checksums()` is redundant as checksums are validated by default. \
1095 For unbuffered resumable uploads, call `with_checksum_precomputation(false)` \
1096 to prioritize speed over upfront integrity validation."
1097 )]
1098 pub async fn precompute_checksums(mut self) -> Result<Self> {
1099 let mut offset = 0_u64;
1100 self.payload.seek(offset).await.map_err(Error::ser)?;
1101 while let Some(n) = self.payload.next().await.transpose().map_err(Error::ser)? {
1102 self.options.checksum.update(offset, &n);
1103 offset += n.len() as u64;
1104 }
1105 self.payload.seek(0_u64).await.map_err(Error::ser)?;
1106 let computed = self.options.checksum.finalize();
1107 let current = self.mut_resource().checksums.get_or_insert_default();
1108 checksum_update(current, computed);
1109 self.options.checksum = Checksum {
1110 crc32c: None,
1111 md5_hash: None,
1112 };
1113 Ok(self)
1114 }
1115}
1116
1117impl<T, S> WriteObject<T, S>
1118where
1119 T: StreamingSource + Send + Sync + 'static,
1120 T::Error: std::error::Error + Send + Sync + 'static,
1121 S: crate::storage::stub::Storage + 'static,
1122{
1123 /// Upload an object from a streaming source without rewinds.
1124 ///
1125 /// If the data source does **not** implement [Seek] the client library must
1126 /// buffer data sent to the service until the service confirms it has
1127 /// persisted the data. This requires more memory in the client, and when
1128 /// the buffer grows too large, may require stalling the writer until the
1129 /// service can persist the data.
1130 ///
1131 /// Use this function for data sources where it is expensive or impossible
1132 /// to restart the data source. This function is also useful when it is hard
1133 /// or impossible to predict the number of bytes emitted by a stream, even
1134 /// if restarting the stream is not too expensive.
1135 ///
1136 /// # Example
1137 /// ```
1138 /// # use google_cloud_storage::client::Storage;
1139 /// # async fn sample(client: &Storage) -> anyhow::Result<()> {
1140 /// let response = client
1141 /// .write_object("projects/_/buckets/my-bucket", "my-object", "hello world")
1142 /// .send_buffered()
1143 /// .await?;
1144 /// println!("response details={response:?}");
1145 /// # Ok(()) }
1146 /// ```
1147 pub async fn send_buffered(self) -> crate::Result<Object> {
1148 self.stub
1149 .write_object_buffered(self.payload, self.request, self.options)
1150 .await
1151 }
1152}
1153
1154// We need `Debug` to use `expect_err()` in `Result<WriteObject, ...>`.
1155impl<T, S> std::fmt::Debug for WriteObject<T, S>
1156where
1157 S: crate::storage::stub::Storage + 'static,
1158{
1159 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1160 f.debug_struct("WriteObject")
1161 .field("stub", &self.stub)
1162 .field("request", &self.request)
1163 // skip payload, as it is not `Debug`
1164 .field("options", &self.options)
1165 .finish()
1166 }
1167}
1168
1169#[cfg(test)]
1170#[allow(deprecated)]
1171mod tests {
1172 use super::client::tests::{test_builder, test_inner_client};
1173 use super::*;
1174 use crate::client::Storage;
1175 use crate::model::{
1176 CommonObjectRequestParams, ObjectChecksums, ObjectContexts, ObjectCustomContextPayload,
1177 WriteObjectSpec,
1178 };
1179 use crate::storage::checksum::details::{Crc32c, Md5};
1180 use crate::streaming_source::tests::MockSeekSource;
1181 use google_cloud_auth::credentials::anonymous::Builder as Anonymous;
1182 use httptest::{Expectation, Server, matchers::*, responders::status_code};
1183 use std::error::Error as _;
1184 use std::io::{Error as IoError, ErrorKind};
1185
1186 type Result = anyhow::Result<()>;
1187
1188 // Verify `write_object()` can be used with a source that implements
1189 // `StreamingSource` **and** `Seek`
1190 #[tokio::test]
1191 async fn test_upload_streaming_source_and_seek() -> Result {
1192 struct Source;
1193 impl crate::streaming_source::StreamingSource for Source {
1194 type Error = std::io::Error;
1195 async fn next(&mut self) -> Option<std::result::Result<bytes::Bytes, Self::Error>> {
1196 None
1197 }
1198 }
1199 impl crate::streaming_source::Seek for Source {
1200 type Error = std::io::Error;
1201 async fn seek(&mut self, _offset: u64) -> std::result::Result<(), Self::Error> {
1202 Ok(())
1203 }
1204 }
1205
1206 let client = Storage::builder()
1207 .with_credentials(Anonymous::new().build())
1208 .build()
1209 .await?;
1210 let _ = client.write_object("projects/_/buckets/test-bucket", "test-object", Source);
1211 Ok(())
1212 }
1213
1214 // Verify `write_object()` can be used with a source that **only**
1215 // implements `StreamingSource`.
1216 #[tokio::test]
1217 async fn test_upload_only_streaming_source() -> Result {
1218 struct Source;
1219 impl crate::streaming_source::StreamingSource for Source {
1220 type Error = std::io::Error;
1221 async fn next(&mut self) -> Option<std::result::Result<bytes::Bytes, Self::Error>> {
1222 None
1223 }
1224 }
1225
1226 let client = Storage::builder()
1227 .with_credentials(Anonymous::new().build())
1228 .build()
1229 .await?;
1230 let _ = client.write_object("projects/_/buckets/test-bucket", "test-object", Source);
1231 Ok(())
1232 }
1233
1234 // Verify `write_object()` meets normal Send, Sync, requirements.
1235 #[tokio::test]
1236 async fn test_upload_is_send_and_static() -> Result {
1237 let client = Storage::builder()
1238 .with_credentials(Anonymous::new().build())
1239 .build()
1240 .await?;
1241
1242 fn need_send<T: Send>(_val: &T) {}
1243 fn need_sync<T: Sync>(_val: &T) {}
1244 fn need_static<T: 'static>(_val: &T) {}
1245
1246 let upload = client.write_object("projects/_/buckets/test-bucket", "test-object", "");
1247 need_send(&upload);
1248 need_sync(&upload);
1249 need_static(&upload);
1250
1251 let upload = client
1252 .write_object("projects/_/buckets/test-bucket", "test-object", "")
1253 .send_unbuffered();
1254 need_send(&upload);
1255 need_static(&upload);
1256
1257 let upload = client
1258 .write_object("projects/_/buckets/test-bucket", "test-object", "")
1259 .send_buffered();
1260 need_send(&upload);
1261 need_static(&upload);
1262
1263 Ok(())
1264 }
1265
1266 #[tokio::test]
1267 async fn write_object_metadata() -> Result {
1268 use crate::model::ObjectAccessControl;
1269 let inner = test_inner_client(test_builder()).await;
1270 let options = inner.options.clone();
1271 let stub = crate::storage::transport::Storage::new_test(inner);
1272 let key = KeyAes256::new(&[0x42; 32]).expect("hard-coded key is not an error");
1273 let mut builder =
1274 WriteObject::new(stub, "projects/_/buckets/bucket", "object", "", options)
1275 .set_if_generation_match(10)
1276 .set_if_generation_not_match(20)
1277 .set_if_metageneration_match(30)
1278 .set_if_metageneration_not_match(40)
1279 .set_predefined_acl("private")
1280 .set_acl([ObjectAccessControl::new()
1281 .set_entity("allAuthenticatedUsers")
1282 .set_role("READER")])
1283 .set_cache_control("public; max-age=7200")
1284 .set_content_disposition("inline")
1285 .set_content_encoding("gzip")
1286 .set_content_language("en")
1287 .set_content_type("text/plain")
1288 .set_contexts(ObjectContexts::new().set_custom([(
1289 "context-key",
1290 ObjectCustomContextPayload::new().set_value("context-value"),
1291 )]))
1292 .set_custom_time(wkt::Timestamp::try_from("2025-07-07T18:11:00Z")?)
1293 .set_event_based_hold(true)
1294 .set_key(key.clone())
1295 .set_metadata([("k0", "v0"), ("k1", "v1")])
1296 .set_retention(
1297 crate::model::object::Retention::new()
1298 .set_mode(crate::model::object::retention::Mode::Locked)
1299 .set_retain_until_time(wkt::Timestamp::try_from("2035-07-07T18:14:00Z")?),
1300 )
1301 .set_storage_class("ARCHIVE")
1302 .set_temporary_hold(true)
1303 .set_kms_key("test-key")
1304 .with_known_crc32c(crc32c::crc32c(b""))
1305 .with_known_md5_hash(md5::compute(b"").0);
1306
1307 let resource = builder.request.spec.resource.take().unwrap();
1308 let builder = builder;
1309 assert_eq!(
1310 &builder.request.spec,
1311 &WriteObjectSpec::new()
1312 .set_if_generation_match(10)
1313 .set_if_generation_not_match(20)
1314 .set_if_metageneration_match(30)
1315 .set_if_metageneration_not_match(40)
1316 .set_predefined_acl("private")
1317 );
1318
1319 assert_eq!(
1320 &builder.request.params,
1321 &Some(CommonObjectRequestParams::from(key))
1322 );
1323
1324 assert_eq!(
1325 resource,
1326 Object::new()
1327 .set_name("object")
1328 .set_bucket("projects/_/buckets/bucket")
1329 .set_acl([ObjectAccessControl::new()
1330 .set_entity("allAuthenticatedUsers")
1331 .set_role("READER")])
1332 .set_cache_control("public; max-age=7200")
1333 .set_content_disposition("inline")
1334 .set_content_encoding("gzip")
1335 .set_content_language("en")
1336 .set_content_type("text/plain")
1337 .set_contexts(ObjectContexts::new().set_custom([(
1338 "context-key",
1339 ObjectCustomContextPayload::new().set_value("context-value"),
1340 )]))
1341 .set_checksums(
1342 crate::model::ObjectChecksums::new()
1343 .set_crc32c(crc32c::crc32c(b""))
1344 .set_md5_hash(bytes::Bytes::from_iter(md5::compute(b"").0))
1345 )
1346 .set_custom_time(wkt::Timestamp::try_from("2025-07-07T18:11:00Z")?)
1347 .set_event_based_hold(true)
1348 .set_metadata([("k0", "v0"), ("k1", "v1")])
1349 .set_retention(
1350 crate::model::object::Retention::new()
1351 .set_mode("LOCKED")
1352 .set_retain_until_time(wkt::Timestamp::try_from("2035-07-07T18:14:00Z")?)
1353 )
1354 .set_storage_class("ARCHIVE")
1355 .set_temporary_hold(true)
1356 .set_kms_key("test-key")
1357 );
1358
1359 Ok(())
1360 }
1361
1362 #[tokio::test]
1363 async fn upload_object_options() {
1364 let inner = test_inner_client(
1365 test_builder()
1366 .with_resumable_upload_threshold(123_usize)
1367 .with_resumable_upload_buffer_size(234_usize),
1368 )
1369 .await;
1370 let options = inner.options.clone();
1371 let stub = crate::storage::transport::Storage::new_test(inner);
1372 let request = WriteObject::new(
1373 stub.clone(),
1374 "projects/_/buckets/bucket",
1375 "object",
1376 "",
1377 options.clone(),
1378 );
1379 assert_eq!(request.options.resumable_upload_threshold(), 123);
1380 assert_eq!(request.options.resumable_upload_buffer_size(), 234);
1381 assert_eq!(request.options.user_agent, None);
1382
1383 let user_agent = "quick_foxes_lazy_dogs/1.0.0";
1384 let request = WriteObject::new(stub, "projects/_/buckets/bucket", "object", "", options)
1385 .with_resumable_upload_threshold(345_usize)
1386 .with_resumable_upload_buffer_size(456_usize)
1387 .with_user_agent(user_agent);
1388 assert_eq!(request.options.resumable_upload_threshold(), 345);
1389 assert_eq!(request.options.resumable_upload_buffer_size(), 456);
1390 assert_eq!(request.options.user_agent.as_deref(), Some(user_agent));
1391 }
1392
1393 const QUICK: &str = "the quick brown fox jumps over the lazy dog";
1394 const VEXING: &str = "how vexingly quick daft zebras jump";
1395
1396 fn quick_checksum(mut engine: Checksum) -> ObjectChecksums {
1397 engine.update(0, &bytes::Bytes::from_static(QUICK.as_bytes()));
1398 engine.finalize()
1399 }
1400
1401 async fn collect<S: StreamingSource>(mut stream: S) -> anyhow::Result<Vec<u8>> {
1402 let mut collected = Vec::new();
1403 while let Some(b) = stream.next().await.transpose()? {
1404 collected.extend_from_slice(&b);
1405 }
1406 Ok(collected)
1407 }
1408
1409 #[tokio::test]
1410 async fn checksum_default() -> Result {
1411 let client = test_builder().build().await?;
1412 let upload = client
1413 .write_object("my-bucket", "my-object", QUICK)
1414 .precompute_checksums()
1415 .await?;
1416 let want = quick_checksum(Checksum {
1417 crc32c: Some(Crc32c::default()),
1418 md5_hash: None,
1419 });
1420 assert_eq!(
1421 upload.request.spec.resource.and_then(|r| r.checksums),
1422 Some(want)
1423 );
1424 let collected = collect(upload.payload).await?;
1425 assert_eq!(collected, QUICK.as_bytes());
1426 Ok(())
1427 }
1428
1429 #[tokio::test]
1430 async fn checksum_md5_and_crc32c() -> Result {
1431 let client = test_builder().build().await?;
1432 let upload = client
1433 .write_object("my-bucket", "my-object", QUICK)
1434 .compute_md5()
1435 .precompute_checksums()
1436 .await?;
1437 let want = quick_checksum(Checksum {
1438 crc32c: Some(Crc32c::default()),
1439 md5_hash: Some(Md5::default()),
1440 });
1441 assert_eq!(
1442 upload.request.spec.resource.and_then(|r| r.checksums),
1443 Some(want)
1444 );
1445 Ok(())
1446 }
1447
1448 #[tokio::test]
1449 async fn checksum_precomputed() -> Result {
1450 let mut engine = Checksum {
1451 crc32c: Some(Crc32c::default()),
1452 md5_hash: Some(Md5::default()),
1453 };
1454 engine.update(0, &bytes::Bytes::from_static(VEXING.as_bytes()));
1455 let ck = engine.finalize();
1456
1457 let client = test_builder().build().await?;
1458 let upload = client
1459 .write_object("my-bucket", "my-object", QUICK)
1460 .with_known_crc32c(ck.crc32c.unwrap())
1461 .with_known_md5_hash(ck.md5_hash.clone())
1462 .precompute_checksums()
1463 .await?;
1464 // Note that the checksums do not match the data. This is intentional,
1465 // we are trying to verify that whatever is provided in with_crc32c()
1466 // and with_md5() is respected.
1467 assert_eq!(
1468 upload.request.spec.resource.and_then(|r| r.checksums),
1469 Some(ck)
1470 );
1471
1472 Ok(())
1473 }
1474
1475 #[tokio::test]
1476 async fn checksum_crc32c_known_md5_computed() -> Result {
1477 let mut engine = Checksum {
1478 crc32c: Some(Crc32c::default()),
1479 md5_hash: Some(Md5::default()),
1480 };
1481 engine.update(0, &bytes::Bytes::from_static(VEXING.as_bytes()));
1482 let ck = engine.finalize();
1483
1484 let client = test_builder().build().await?;
1485 let upload = client
1486 .write_object("my-bucket", "my-object", QUICK)
1487 .compute_md5()
1488 .with_known_crc32c(ck.crc32c.unwrap())
1489 .precompute_checksums()
1490 .await?;
1491 // Note that the checksums do not match the data. This is intentional,
1492 // we are trying to verify that whatever is provided in with_known*()
1493 // is respected.
1494 let want = quick_checksum(Checksum {
1495 crc32c: None,
1496 md5_hash: Some(Md5::default()),
1497 })
1498 .set_crc32c(ck.crc32c.unwrap());
1499 assert_eq!(
1500 upload.request.spec.resource.and_then(|r| r.checksums),
1501 Some(want)
1502 );
1503
1504 Ok(())
1505 }
1506
1507 #[tokio::test]
1508 async fn checksum_mixed_then_precomputed() -> Result {
1509 let mut engine = Checksum {
1510 crc32c: Some(Crc32c::default()),
1511 md5_hash: Some(Md5::default()),
1512 };
1513 engine.update(0, &bytes::Bytes::from_static(VEXING.as_bytes()));
1514 let ck = engine.finalize();
1515
1516 let client = test_builder().build().await?;
1517 let upload = client
1518 .write_object("my-bucket", "my-object", QUICK)
1519 .with_known_md5_hash(ck.md5_hash.clone())
1520 .with_known_crc32c(ck.crc32c.unwrap())
1521 .precompute_checksums()
1522 .await?;
1523 // Note that the checksums do not match the data. This is intentional,
1524 // we are trying to verify that whatever is provided in with_known*()
1525 // is respected.
1526 let want = ck.clone();
1527 assert_eq!(
1528 upload.request.spec.resource.and_then(|r| r.checksums),
1529 Some(want)
1530 );
1531
1532 Ok(())
1533 }
1534
1535 #[tokio::test]
1536 async fn checksum_full_computed_then_md5_precomputed() -> Result {
1537 let mut engine = Checksum {
1538 crc32c: Some(Crc32c::default()),
1539 md5_hash: Some(Md5::default()),
1540 };
1541 engine.update(0, &bytes::Bytes::from_static(VEXING.as_bytes()));
1542 let ck = engine.finalize();
1543
1544 let client = test_builder().build().await?;
1545 let upload = client
1546 .write_object("my-bucket", "my-object", QUICK)
1547 .compute_md5()
1548 .with_known_md5_hash(ck.md5_hash.clone())
1549 .precompute_checksums()
1550 .await?;
1551 // Note that the checksums do not match the data. This is intentional,
1552 // we are trying to verify that whatever is provided in with_known*()
1553 // is respected.
1554 let want = quick_checksum(Checksum {
1555 crc32c: Some(Crc32c::default()),
1556 md5_hash: None,
1557 })
1558 .set_md5_hash(ck.md5_hash.clone());
1559 assert_eq!(
1560 upload.request.spec.resource.and_then(|r| r.checksums),
1561 Some(want)
1562 );
1563
1564 Ok(())
1565 }
1566
1567 #[tokio::test]
1568 async fn checksum_known_crc32_then_computed_md5() -> Result {
1569 let mut engine = Checksum {
1570 crc32c: Some(Crc32c::default()),
1571 md5_hash: Some(Md5::default()),
1572 };
1573 engine.update(0, &bytes::Bytes::from_static(VEXING.as_bytes()));
1574 let ck = engine.finalize();
1575
1576 let client = test_builder().build().await?;
1577 let upload = client
1578 .write_object("my-bucket", "my-object", QUICK)
1579 .with_known_crc32c(ck.crc32c.unwrap())
1580 .compute_md5()
1581 .with_known_md5_hash(ck.md5_hash.clone())
1582 .precompute_checksums()
1583 .await?;
1584 // Note that the checksums do not match the data. This is intentional,
1585 // we are trying to verify that whatever is provided in with_known*()
1586 // is respected.
1587 let want = ck.clone();
1588 assert_eq!(
1589 upload.request.spec.resource.and_then(|r| r.checksums),
1590 Some(want)
1591 );
1592
1593 Ok(())
1594 }
1595
1596 #[tokio::test]
1597 async fn checksum_known_crc32_then_known_md5() -> Result {
1598 let mut engine = Checksum {
1599 crc32c: Some(Crc32c::default()),
1600 md5_hash: Some(Md5::default()),
1601 };
1602 engine.update(0, &bytes::Bytes::from_static(VEXING.as_bytes()));
1603 let ck = engine.finalize();
1604
1605 let client = test_builder().build().await?;
1606 let upload = client
1607 .write_object("my-bucket", "my-object", QUICK)
1608 .with_known_crc32c(ck.crc32c.unwrap())
1609 .with_known_md5_hash(ck.md5_hash.clone())
1610 .precompute_checksums()
1611 .await?;
1612 // Note that the checksums do not match the data. This is intentional,
1613 // we are trying to verify that whatever is provided in with_known*()
1614 // is respected.
1615 let want = ck.clone();
1616 assert_eq!(
1617 upload.request.spec.resource.and_then(|r| r.checksums),
1618 Some(want)
1619 );
1620
1621 Ok(())
1622 }
1623
1624 #[tokio::test]
1625 async fn precompute_checksums_seek_error() -> Result {
1626 let mut source = MockSeekSource::new();
1627 source
1628 .expect_seek()
1629 .once()
1630 .returning(|_| Err(IoError::new(ErrorKind::Deadlock, "test-only")));
1631
1632 let client = test_builder().build().await?;
1633 let err = client
1634 .write_object("my-bucket", "my-object", source)
1635 .precompute_checksums()
1636 .await
1637 .expect_err("seek() returns an error");
1638 assert!(err.is_serialization(), "{err:?}");
1639 assert!(
1640 err.source()
1641 .and_then(|e| e.downcast_ref::<IoError>())
1642 .is_some(),
1643 "{err:?}"
1644 );
1645
1646 Ok(())
1647 }
1648
1649 #[tokio::test]
1650 async fn precompute_checksums_next_error() -> Result {
1651 let mut source = MockSeekSource::new();
1652 source.expect_seek().returning(|_| Ok(()));
1653 let mut seq = mockall::Sequence::new();
1654 source
1655 .expect_next()
1656 .times(3)
1657 .in_sequence(&mut seq)
1658 .returning(|| Some(Ok(bytes::Bytes::new())));
1659 source
1660 .expect_next()
1661 .once()
1662 .in_sequence(&mut seq)
1663 .returning(|| Some(Err(IoError::new(ErrorKind::BrokenPipe, "test-only"))));
1664
1665 let client = test_builder().build().await?;
1666 let err = client
1667 .write_object("my-bucket", "my-object", source)
1668 .precompute_checksums()
1669 .await
1670 .expect_err("seek() returns an error");
1671 assert!(err.is_serialization(), "{err:?}");
1672 assert!(
1673 err.source()
1674 .and_then(|e| e.downcast_ref::<IoError>())
1675 .is_some(),
1676 "{err:?}"
1677 );
1678
1679 Ok(())
1680 }
1681
1682 #[tokio::test]
1683 async fn write_object_with_user_agent() -> Result {
1684 use http::header::USER_AGENT;
1685
1686 let user_agent = "quicker_foxes_lazier_dogs/1.2.3";
1687 let server = Server::run();
1688 server.expect(
1689 Expectation::matching(all_of![
1690 request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
1691 request::headers(contains((USER_AGENT.as_str(), user_agent))),
1692 request::query(url_decoded(contains(("uploadType", "multipart")))),
1693 ])
1694 .times(1)
1695 .respond_with(status_code(200).body("{}")),
1696 );
1697
1698 let client = Storage::builder()
1699 .with_endpoint(format!("http://{}", server.addr()))
1700 .with_credentials(Anonymous::new().build())
1701 .build()
1702 .await?;
1703 let _ = client
1704 .write_object(
1705 "projects/_/buckets/test-bucket",
1706 "test-object",
1707 "hello world",
1708 )
1709 .with_user_agent(user_agent)
1710 .send_unbuffered()
1711 .await?;
1712
1713 Ok(())
1714 }
1715
1716 #[tokio::test]
1717 async fn write_object_with_quota_project() -> Result {
1718 const PROJECT_NAME: &str = "project_lazy_dog";
1719 let server = Server::run();
1720 let session = server.url("/upload/session/test-only-001");
1721 let path = session.path().to_string();
1722
1723 server.expect(
1724 Expectation::matching(all_of![
1725 request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
1726 request::headers(contains(("x-goog-user-project", PROJECT_NAME))),
1727 request::query(url_decoded(contains(("uploadType", "resumable")))),
1728 ])
1729 .times(1)
1730 .respond_with(status_code(200).append_header("location", session.to_string())),
1731 );
1732
1733 server.expect(
1734 Expectation::matching(all_of![
1735 request::method_path("PUT", path),
1736 request::headers(contains(("x-goog-user-project", PROJECT_NAME))),
1737 ])
1738 .times(1)
1739 .respond_with(
1740 status_code(200)
1741 .append_header(http::header::CONTENT_TYPE, "application/json")
1742 .body(serde_json::to_string(&crate::model::Object::new()).unwrap()),
1743 ),
1744 );
1745
1746 let client = Storage::builder()
1747 .with_endpoint(format!("http://{}", server.addr()))
1748 .with_credentials(Anonymous::new().build())
1749 .build()
1750 .await?;
1751 let _ = client
1752 .write_object(
1753 "projects/_/buckets/test-bucket",
1754 "test-object",
1755 "hello world",
1756 )
1757 .with_quota_project(PROJECT_NAME)
1758 .with_resumable_upload_threshold(0_usize)
1759 .send_unbuffered()
1760 .await?;
1761
1762 Ok(())
1763 }
1764
1765 #[tokio::test]
1766 async fn debug() -> Result {
1767 let client = test_builder().build().await?;
1768 let upload = client
1769 .write_object("my-bucket", "my-object", "")
1770 .precompute_checksums()
1771 .await;
1772
1773 let fmt = format!("{upload:?}");
1774 ["WriteObject", "inner", "spec", "options", "checksum"]
1775 .into_iter()
1776 .for_each(|text| {
1777 assert!(fmt.contains(text), "expected {text} in {fmt}");
1778 });
1779 Ok(())
1780 }
1781
1782 #[tokio::test]
1783 async fn with_checksum_precomputation_builder() -> Result {
1784 let client = test_builder().build().await?;
1785 let upload = client
1786 .write_object("my-bucket", "my-object", "hello")
1787 .with_checksum_precomputation(true);
1788 assert!(upload.request.checksum_precomputation);
1789
1790 let upload = client
1791 .write_object("my-bucket", "my-object", "hello")
1792 .with_checksum_precomputation(false);
1793 assert!(!upload.request.checksum_precomputation);
1794
1795 let upload = client.write_object("my-bucket", "my-object", "hello");
1796 assert!(upload.request.checksum_precomputation);
1797 Ok(())
1798 }
1799
1800 #[tokio::test]
1801 async fn unbuffered_resumable_precomputes_by_default() -> Result {
1802 let server = Server::run();
1803 let session = server.url("/upload/session/test-only-001");
1804 let path = session.path().to_string();
1805
1806 use base64::{Engine, prelude::BASE64_STANDARD};
1807 let expected_crc = crc32c::crc32c(b"hello world");
1808 let expected_crc_b64 = BASE64_STANDARD.encode(expected_crc.to_be_bytes());
1809
1810 server.expect(
1811 Expectation::matching(all_of![
1812 request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
1813 request::query(url_decoded(contains(("uploadType", "resumable")))),
1814 request::body(json_decoded(move |body: &serde_json::Value| {
1815 body.get("crc32c").and_then(|v| v.as_str()) == Some(&expected_crc_b64)
1816 })),
1817 ])
1818 .times(1)
1819 .respond_with(status_code(200).append_header("location", session.to_string())),
1820 );
1821
1822 server.expect(
1823 Expectation::matching(all_of![
1824 request::method_path("PUT", path),
1825 request::body("hello world"),
1826 ])
1827 .times(1)
1828 .respond_with(
1829 status_code(200)
1830 .append_header(http::header::CONTENT_TYPE, "application/json")
1831 .body(serde_json::to_string(&crate::model::Object::new()).unwrap()),
1832 ),
1833 );
1834
1835 let client = Storage::builder()
1836 .with_endpoint(format!("http://{}", server.addr()))
1837 .with_credentials(Anonymous::new().build())
1838 .build()
1839 .await?;
1840 let _ = client
1841 .write_object(
1842 "projects/_/buckets/test-bucket",
1843 "test-object",
1844 "hello world",
1845 )
1846 .with_resumable_upload_threshold(0_usize)
1847 .send_unbuffered()
1848 .await?;
1849
1850 Ok(())
1851 }
1852
1853 #[tokio::test]
1854 async fn unbuffered_resumable_with_checksum_precomputation_false() -> Result {
1855 let server = Server::run();
1856 let session = server.url("/upload/session/test-only-002");
1857 let path = session.path().to_string();
1858
1859 server.expect(
1860 Expectation::matching(all_of![
1861 request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
1862 request::query(url_decoded(contains(("uploadType", "resumable")))),
1863 request::body(json_decoded(|body: &serde_json::Value| {
1864 body.get("crc32c").is_none()
1865 })),
1866 ])
1867 .times(1)
1868 .respond_with(status_code(200).append_header("location", session.to_string())),
1869 );
1870
1871 server.expect(
1872 Expectation::matching(request::method_path("PUT", path))
1873 .times(1)
1874 .respond_with(
1875 status_code(200)
1876 .append_header(http::header::CONTENT_TYPE, "application/json")
1877 .body(serde_json::to_string(&crate::model::Object::new()).unwrap()),
1878 ),
1879 );
1880
1881 let client = Storage::builder()
1882 .with_endpoint(format!("http://{}", server.addr()))
1883 .with_credentials(Anonymous::new().build())
1884 .build()
1885 .await?;
1886 let _ = client
1887 .write_object(
1888 "projects/_/buckets/test-bucket",
1889 "test-object",
1890 "hello world",
1891 )
1892 .with_resumable_upload_threshold(0_usize)
1893 .with_checksum_precomputation(false)
1894 .send_unbuffered()
1895 .await?;
1896
1897 Ok(())
1898 }
1899
1900 #[tokio::test]
1901 async fn unbuffered_single_shot_does_not_precompute_by_default() -> Result {
1902 let server = Server::run();
1903 server.expect(
1904 Expectation::matching(all_of![
1905 request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
1906 request::query(url_decoded(contains(("uploadType", "multipart")))),
1907 ])
1908 .times(1)
1909 .respond_with(
1910 status_code(200)
1911 .append_header(http::header::CONTENT_TYPE, "application/json")
1912 .body(serde_json::to_string(&crate::model::Object::new()).unwrap()),
1913 ),
1914 );
1915
1916 let client = Storage::builder()
1917 .with_endpoint(format!("http://{}", server.addr()))
1918 .with_credentials(Anonymous::new().build())
1919 .build()
1920 .await?;
1921 let _ = client
1922 .write_object(
1923 "projects/_/buckets/test-bucket",
1924 "test-object",
1925 "hello world",
1926 )
1927 .send_unbuffered()
1928 .await?;
1929
1930 Ok(())
1931 }
1932
1933 #[tokio::test]
1934 async fn unbuffered_single_shot_with_checksum_precomputation_false() -> Result {
1935 let server = Server::run();
1936 server.expect(
1937 Expectation::matching(all_of![
1938 request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
1939 request::query(url_decoded(contains(("uploadType", "multipart")))),
1940 ])
1941 .times(1)
1942 .respond_with(
1943 status_code(200)
1944 .append_header(http::header::CONTENT_TYPE, "application/json")
1945 .body(serde_json::to_string(&crate::model::Object::new()).unwrap()),
1946 ),
1947 );
1948
1949 let client = Storage::builder()
1950 .with_endpoint(format!("http://{}", server.addr()))
1951 .with_credentials(Anonymous::new().build())
1952 .build()
1953 .await?;
1954 let _ = client
1955 .write_object(
1956 "projects/_/buckets/test-bucket",
1957 "test-object",
1958 "hello world",
1959 )
1960 .with_checksum_precomputation(false)
1961 .send_unbuffered()
1962 .await?;
1963
1964 Ok(())
1965 }
1966
1967 #[tokio::test]
1968 async fn unbuffered_resumable_with_known_crc32c_skips_precompute() -> Result {
1969 let server = Server::run();
1970 let session = server.url("/upload/session/test-only-003");
1971 let path = session.path().to_string();
1972
1973 let known_crc = 123456_u32;
1974 use base64::{Engine, prelude::BASE64_STANDARD};
1975 let expected_crc_b64 = BASE64_STANDARD.encode(known_crc.to_be_bytes());
1976
1977 server.expect(
1978 Expectation::matching(all_of![
1979 request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
1980 request::query(url_decoded(contains(("uploadType", "resumable")))),
1981 request::body(json_decoded(move |body: &serde_json::Value| {
1982 body.get("crc32c").and_then(|v| v.as_str()) == Some(&expected_crc_b64)
1983 })),
1984 ])
1985 .times(1)
1986 .respond_with(status_code(200).append_header("location", session.to_string())),
1987 );
1988
1989 server.expect(
1990 Expectation::matching(request::method_path("PUT", path))
1991 .times(1)
1992 .respond_with(
1993 status_code(200)
1994 .append_header(http::header::CONTENT_TYPE, "application/json")
1995 .body(serde_json::to_string(&crate::model::Object::new()).unwrap()),
1996 ),
1997 );
1998
1999 let client = Storage::builder()
2000 .with_endpoint(format!("http://{}", server.addr()))
2001 .with_credentials(Anonymous::new().build())
2002 .build()
2003 .await?;
2004 let _ = client
2005 .write_object(
2006 "projects/_/buckets/test-bucket",
2007 "test-object",
2008 "hello world",
2009 )
2010 .with_known_crc32c(known_crc)
2011 .with_resumable_upload_threshold(0_usize)
2012 .send_unbuffered()
2013 .await?;
2014
2015 Ok(())
2016 }
2017}