Merge branch 'fix/content_type' into bool-api

ecstore/src/options.rs

@@ -1,12 +1,14 @@
-use http::{HeaderMap, HeaderValue};
-use uuid::Uuid;
-
+use crate::bucket::metadata;
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::error::{Error, Result};
 use crate::store_api::ObjectOptions;
 use crate::store_err::StorageError;
 use crate::utils::path::is_dir_object;
+use http::{HeaderMap, HeaderValue};
+use lazy_static::lazy_static;
 use std::collections::HashMap;
+use tracing::warn;
+use uuid::Uuid;
 
 pub async fn put_opts(
     bucket: &str,
@@ -58,14 +60,67 @@ pub fn put_opts_from_headers(
     headers: &HeaderMap<HeaderValue>,
     metadata: Option<HashMap<String, String>>,
 ) -> Result<ObjectOptions> {
-    // TODO custom headers
+    let metadata = metadata.unwrap_or_default();
+
     get_default_opts(headers, metadata, false)
 }
 
 fn get_default_opts(
-    _headers: &HeaderMap<HeaderValue>,
-    _metadata: Option<HashMap<String, String>>,
+    headers: &HeaderMap<HeaderValue>,
+    metadata: HashMap<String, String>,
     _copy_source: bool,
 ) -> Result<ObjectOptions> {
-    Ok(ObjectOptions::default())
+    warn!("get headers: {:?}", &headers);
+    warn!("get metadata: {:?}", &metadata);
+
+    Ok(ObjectOptions {
+        user_defined: metadata.clone(),
+        ..Default::default()
+    })
+}
+
+pub fn extract_metadata(headers: &HeaderMap<HeaderValue>) -> HashMap<String, String> {
+    let mut metadata = HashMap::new();
+
+    extract_metadata_from_mime(&headers, &mut metadata);
+
+    metadata
+}
+
+fn extract_metadata_from_mime(headers: &HeaderMap<HeaderValue>, metadata: &mut HashMap<String, String>) {
+    for (k, v) in headers.iter() {
+        if k.as_str().starts_with("x-amz-meta-") {
+            metadata.insert(k.to_string(), String::from_utf8_lossy(v.as_bytes()).to_string());
+            continue;
+        }
+
+        if k.as_str().starts_with("x-rustfs-meta-") {
+            metadata.insert(k.to_string(), String::from_utf8_lossy(v.as_bytes()).to_string());
+            continue;
+        }
+
+        for hd in SUPPORTED_HEADERS.iter() {
+            if k.as_str() == *hd {
+                metadata.insert(k.to_string(), String::from_utf8_lossy(v.as_bytes()).to_string());
+                continue;
+            }
+        }
+    }
+
+    if !metadata.contains_key("content-type") {
+        metadata.insert("content-type".to_owned(), "binary/octet-stream".to_owned());
+    }
+}
+lazy_static! {
+    static ref SUPPORTED_HEADERS: Vec<&'static str> = vec![
+        "content-type",
+        "cache-control",
+        "content-language",
+        "content-encoding",
+        "content-disposition",
+        "x-amz-storage-class",
+        "x-amz-tagging",
+        "expires",
+        "x-amz-replication-status"
+    ];
 }

ecstore/src/store_api.rs

@@ -537,6 +537,8 @@ pub struct ObjectOptions {
 
     pub data_movement: bool,
     pub src_pool_idx: usize,
+    pub user_defined: HashMap<String, String>,
+    pub preserve_etag: Option<String>,
 }
 
 // impl Default for ObjectOptions {

ecstore/src/xhttp.rs

@@ -1 +1,2 @@
-pub(crate) const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
+pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
+pub const AMZ_STORAGE_CLASS: &str = "x-amz-storage-class";

rustfs/src/storage/ecfs.rs

@@ -18,6 +18,7 @@ use ecstore::bucket::tagging::encode_tags;
 use ecstore::bucket::versioning_sys::BucketVersioningSys;
 use ecstore::disk::error::DiskError;
 use ecstore::new_object_layer_fn;
+use ecstore::options::extract_metadata;
 use ecstore::options::put_opts;
 use ecstore::store_api::BucketOptions;
 use ecstore::store_api::CompletePart;
@@ -30,6 +31,7 @@ use ecstore::store_api::ObjectOptions;
 use ecstore::store_api::ObjectToDelete;
 use ecstore::store_api::PutObjReader;
 use ecstore::store_api::StorageAPI;
+use ecstore::xhttp;
 use futures::pin_mut;
 use futures::{Stream, StreamExt};
 use http::HeaderMap;
@@ -45,6 +47,7 @@ use s3s::{S3Request, S3Response};
 use std::fmt::Debug;
 use std::str::FromStr;
 use tracing::debug;
+use tracing::error;
 use tracing::info;
 use transform_stream::AsyncTryStream;
 use uuid::Uuid;
@@ -325,14 +328,28 @@ impl S3 for FS {
 
         let info = reader.object_info;
 
-        let content_type = try_!(ContentType::from_str("application/x-msdownload"));
+        let content_type = {
+            if let Some(content_type) = info.content_type {
+                let ct = match ContentType::from_str(&content_type) {
+                    Ok(res) => Some(res),
+                    Err(err) => {
+                        error!("parse content-type err {} {:?}", &content_type, err);
+                        //
+                        None
+                    }
+                };
+                ct
+            } else {
+                None
+            }
+        };
         let last_modified = info.mod_time.map(Timestamp::from);
 
         let output = GetObjectOutput {
             body: Some(reader.stream),
             content_length: Some(info.size as i64),
             last_modified,
-            content_type: Some(content_type),
+            content_type,
             ..Default::default()
         };
 
@@ -378,12 +395,26 @@ impl S3 for FS {
         let info = try_!(store.get_object_info(&bucket, &key, &ObjectOptions::default()).await);
         debug!("info {:?}", info);
 
-        let content_type = try_!(ContentType::from_str("application/x-msdownload"));
+        let content_type = {
+            if let Some(content_type) = info.content_type {
+                let ct = match ContentType::from_str(&content_type) {
+                    Ok(res) => Some(res),
+                    Err(err) => {
+                        error!("parse content-type err {} {:?}", &content_type, err);
+                        //
+                        None
+                    }
+                };
+                ct
+            } else {
+                None
+            }
+        };
         let last_modified = info.mod_time.map(Timestamp::from);
 
         let output = HeadObjectOutput {
             content_length: Some(try_!(i64::try_from(info.size))),
-            content_type: Some(content_type),
+            content_type,
             last_modified,
             // metadata: object_metadata,
             ..Default::default()
@@ -534,8 +565,8 @@ impl S3 for FS {
             body,
             bucket,
             key,
-            metadata,
             content_length,
+            tagging,
             ..
         } = input;
 
@@ -552,7 +583,12 @@ impl S3 for FS {
             None => return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())),
         };
 
-        let opts: ObjectOptions = try_!(put_opts(&bucket, &key, None, &req.headers, metadata).await);
+        let mut metadata = extract_metadata(&req.headers);
+        if let Some(tags) = tagging {
+            metadata.insert(xhttp::AMZ_OBJECT_TAGGING.to_owned(), tags);
+        }
+
+        let opts: ObjectOptions = try_!(put_opts(&bucket, &key, None, &req.headers, Some(metadata)).await);
 
         let obj_info = try_!(store.put_object(&bucket, &key, &mut reader, &opts).await);
 
@@ -573,12 +609,12 @@ impl S3 for FS {
         req: S3Request<CreateMultipartUploadInput>,
     ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
         let CreateMultipartUploadInput {
-            bucket, key, metadata, ..
+            bucket, key, tagging, ..
         } = req.input;
 
         // mc cp step 3
 
-        debug!("create_multipart_upload meta {:?}", &metadata);
+        // debug!("create_multipart_upload meta {:?}", &metadata);
 
         let layer = new_object_layer_fn();
         let lock = layer.read().await;
@@ -587,8 +623,15 @@ impl S3 for FS {
             None => return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())),
         };
 
-        let MultipartUploadResult { upload_id, .. } =
-            try_!(store.new_multipart_upload(&bucket, &key, &ObjectOptions::default()).await);
+        let mut metadata = extract_metadata(&req.headers);
+
+        if let Some(tags) = tagging {
+            metadata.insert(xhttp::AMZ_OBJECT_TAGGING.to_owned(), tags);
+        }
+
+        let opts: ObjectOptions = try_!(put_opts(&bucket, &key, None, &req.headers, Some(metadata)).await);
+
+        let MultipartUploadResult { upload_id, .. } = try_!(store.new_multipart_upload(&bucket, &key, &opts).await);
 
         let output = CreateMultipartUploadOutput {
             bucket: Some(bucket),
@@ -609,6 +652,7 @@ impl S3 for FS {
             upload_id,
             part_number,
             content_length,
+            // content_md5,
             ..
         } = req.input;
 
@@ -630,6 +674,8 @@ impl S3 for FS {
             None => return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())),
         };
 
+        // TODO: hash_reader
+
         let info = try_!(
             store
                 .put_object_part(&bucket, &key, &upload_id, part_id, &mut data, &opts)