feat(services/gcs): Allow setting default storage_class

core/src/services/gcs/backend.rs

@@ -62,6 +62,7 @@ const DEFAULT_GCS_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read
 /// - `endpoint`: Customizable endpoint setting
 /// - `credentials`: Credential string for GCS OAuth2
 /// - `predefined_acl`: Predefined ACL for GCS
+/// - `default_storage_class`: Default storage class for GCS
 ///
 /// You can refer to [`GcsBuilder`]'s docs for more information
 ///
@@ -88,6 +89,8 @@ const DEFAULT_GCS_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read
 ///     builder.credential("authentication token");
 ///     // set the predefined ACL for GCS
 ///     builder.predefined_acl("publicRead");
+///     // set the default storage class for GCS
+///     builder.default_storage_class("STANDARD");
 ///
 ///     let op: Operator = Operator::new(builder)?.finish();
 ///     Ok(())
@@ -115,6 +118,7 @@ pub struct GcsBuilder {
     http_client: Option<HttpClient>,
     customed_token_loader: Option<Box<dyn GoogleTokenLoad>>,
     predefined_acl: Option<String>,
+    default_storage_class: String,
 }
 
 impl GcsBuilder {
@@ -218,6 +222,20 @@ impl GcsBuilder {
         };
         self
     }
+
+    /// Set the default storage class for GCS.
+    ///
+    /// Available values are:
+    /// - `STANDARD`
+    /// - `NEARLINE`
+    /// - `COLDLINE`
+    /// - `ARCHIVE`
+    pub fn default_storage_class(&mut self, class: &str) -> &mut Self {
+        if !class.is_empty() {
+            self.default_storage_class = class.to_string()
+        };
+        self
+    }
 }
 
 impl Debug for GcsBuilder {
@@ -233,6 +251,7 @@ impl Debug for GcsBuilder {
         if self.predefined_acl.is_some() {
             ds.field("predefined_acl", &self.predefined_acl);
         }
+        ds.field("default_storage_class", &self.default_storage_class);
         ds.finish()
     }
 }
@@ -250,6 +269,8 @@ impl Builder for GcsBuilder {
         map.get("credential").map(|v| builder.credential(v));
         map.get("scope").map(|v| builder.scope(v));
         map.get("predefined_acl").map(|v| builder.predefined_acl(v));
+        map.get("default_storage_class")
+            .map(|v| builder.default_storage_class(v));
 
         builder
     }
@@ -314,6 +335,10 @@ impl Builder for GcsBuilder {
 
         let signer = GoogleSigner::new("storage");
 
+        if self.default_storage_class.is_empty() {
+            self.default_storage_class = "STANDARD".to_string();
+        }
+
         let backend = GcsBackend {
             core: Arc::new(GcsCore {
                 endpoint,
@@ -324,6 +349,7 @@ impl Builder for GcsBuilder {
                 token_loader,
                 credential_loader: cred_loader,
                 predefined_acl: self.predefined_acl.clone(),
+                default_storage_class: self.default_storage_class.clone(),
             }),
         };
 

core/src/services/gcs/core.rs

@@ -21,6 +21,7 @@ use std::fmt::Write;
 
 use backon::ExponentialBuilder;
 use backon::Retryable;
+use bytes::BytesMut;
 use http::header::CONTENT_LENGTH;
 use http::header::CONTENT_TYPE;
 use http::Request;
@@ -46,15 +47,20 @@ pub struct GcsCore {
     pub credential_loader: GoogleCredentialLoader,
 
     pub predefined_acl: Option<String>,
+    pub default_storage_class: String,
 }
 
 impl Debug for GcsCore {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         let mut de = f.debug_struct("Backend");
         de.field("endpoint", &self.endpoint)
             .field("bucket", &self.bucket)
-            .field("root", &self.root)
-            .finish_non_exhaustive()
+            .field("root", &self.root);
+        if self.predefined_acl.is_some() {
+            de.field("predefined_acl", &self.predefined_acl);
+        }
+        de.field("default_storage_class", &self.default_storage_class);
+        de.finish_non_exhaustive()
     }
 }
 
@@ -139,7 +145,7 @@ impl GcsCore {
         let p = build_abs_path(&self.root, path);
 
         let mut url = format!(
-            "{}/upload/storage/v1/b/{}/o?uploadType=media&name={}",
+            "{}/upload/storage/v1/b/{}/o?uploadType=multipart&name={}",
             self.endpoint,
             self.bucket,
             percent_encode_path(&p)
@@ -149,19 +155,32 @@ impl GcsCore {
             write!(&mut url, "&predefinedAcl={}", acl).unwrap();
         }
 
-        let mut req = Request::post(&url);
+        let mut req =
+            Request::post(&url).header(CONTENT_TYPE, "multipart/related; boundary=my-boundary");
 
         if let Some(size) = size {
             req = req.header(CONTENT_LENGTH, size)
         }
 
+        let mut req_body = BytesMut::with_capacity(100);
+        write!(
+            &mut req_body,
+            "--my-boundary\nContent-Type: application/json; charset=UTF-8\n\n{{\"storageClass\": \"{}\"}}\n\n--my-boundary\n",
+            self.default_storage_class
+        ).unwrap();
+
         if let Some(mime) = content_type {
-            req = req.header(CONTENT_TYPE, mime)
+            write!(&mut req_body, "Content-Type: {}\n\n", mime).unwrap();
+        } else {
+            write!(&mut req_body, "Content-Type: application/octet-stream\n\n").unwrap();
         }
 
-        // Set body
-        let req = req.body(body).map_err(new_request_build_error)?;
-
+        if let AsyncBody::Bytes(bytes) = body {
+            req_body.extend_from_slice(&bytes);
+        }
+        write!(&mut req_body, "\n--my-boundary").unwrap();
+        let req_body = AsyncBody::Bytes(req_body.freeze());
+        let req = req.body(req_body).map_err(new_request_build_error)?;
         Ok(req)
     }