ufile_rus3/api/
multipart_finish.rs

1//! This module used to finish the multipart upload task.
2
3use std::{collections::HashMap, fmt::Display};
4
5use anyhow::Error;
6use chrono::Local;
7use reqwest::{
8    Method,
9    header::{HeaderMap, HeaderName},
10};
11
12use crate::{
13    AuthorizationService,
14    api::{
15        ApiOperation, ObjectOptAuthParamBuilder,
16        object::{BaseResponse, FinishUploadResponse, InitMultipartState, MultipartUploadState},
17    },
18    define_api_request, define_operation_struct,
19};
20
21define_operation_struct!(MultipartFinishOperation);
22define_api_request!(
23    MultipartFinishRequest,
24    MultipartFinishOperationBuilder,
25    FinishUploadResponse,
26    {
27        /// Required: Slice initial state
28        pub state: InitMultipartState,
29
30        /// Required: Slice states
31        pub part_states: Vec<MultipartUploadState>,
32
33        /// Optional: new object name used to replace old one if finish multipart upload task successfully.
34        #[builder(setter(into, strip_option), default)]
35        pub new_object: Option<String>,
36
37        /// Optional: UNCHANGED（默认值）:保持初始化时设置的用户自定义元数据不变。
38        ///
39        /// REPLACE：忽略初始化分片时设置的用户自定义元数据，直接采用Finish请求中指定的元数据。
40        #[builder(setter(into, strip_option), default)]
41        pub metadata_directive: Option<MetadataDirective>,
42
43        /// Optional: User custom headers metadata.
44        #[builder(setter(into, strip_option), default)]
45        pub metadata: Option<HashMap<String, String>>,
46
47        /// Optional: Security Token
48        #[builder(setter(into, strip_option), default)]
49        pub security_token: Option<String>,
50    }
51);
52
53/// UNCHANGED（默认值）:保持初始化时设置的用户自定义元数据不变。
54///
55/// REPLACE：忽略初始化分片时设置的用户自定义元数据，直接采用Finish请求中指定的元数据。
56#[allow(unused)]
57#[derive(Debug, Clone, Copy)]
58pub enum MetadataDirective {
59    Unchanged,
60    Replace,
61}
62
63impl Display for MetadataDirective {
64    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        match self {
66            MetadataDirective::Unchanged => write!(f, "UNCHANGED"),
67            MetadataDirective::Replace => write!(f, "REPLACE"),
68        }
69    }
70}
71
72#[async_trait::async_trait]
73impl ApiOperation for MultipartFinishOperation {
74    type Request = MultipartFinishRequest;
75    type Response = FinishUploadResponse;
76    type Error = Error;
77
78    async fn execute(&self, req: Self::Request) -> Result<Self::Response, Self::Error> {
79        let MultipartFinishRequest {
80            state,
81            mut part_states,
82            new_object,
83            metadata_directive,
84            metadata,
85            security_token,
86            ..
87        } = req;
88        let mime_type = state
89            .mime_type
90            .clone()
91            .ok_or(Error::msg("mime type is unset."))?;
92        // let mime_type = "text/plain".to_string();
93        let date = Local::now().format("%Y%m%d%H%M%S").to_string();
94        let auth_object = ObjectOptAuthParamBuilder::default()
95            .method(Method::POST)
96            .bucket(state.bucket.as_str())
97            .key_name(state.key_name.as_str())
98            .content_type(mime_type.as_str())
99            .date(date.as_str())
100            .build()?;
101        let authorization =
102            AuthorizationService.authorization(auth_object, self.object_config.clone())?;
103        let mut headers = HeaderMap::new();
104        headers.insert("Content-Type", mime_type.parse().unwrap());
105        headers.insert("Accept", "*/*".parse().unwrap());
106        headers.insert("Date", date.parse().unwrap());
107        headers.insert("Authorization", authorization.parse().unwrap());
108        if let Some(ref security_token) = security_token
109            && !security_token.is_empty()
110        {
111            headers.insert("SecurityToken", security_token.parse().unwrap());
112        }
113        if let Some(ref directive) = metadata_directive {
114            headers.insert(
115                "X-Ufile-Metadata-Directive",
116                directive.to_string().parse().unwrap(),
117            );
118        }
119        // We must add metadata to headers if metadata is not empty.
120        let url = self
121            .object_config
122            .generate_final_host(state.bucket.as_str(), state.key_name.as_str());
123        let url = format!(
124            "{}?uploadId={}&newKey={}",
125            url,
126            state.upload_id,
127            new_object.as_ref().unwrap_or(&String::new())
128        );
129        // calc body.
130        part_states.sort_by(|a, b| a.part_number.cmp(&b.part_number));
131        let body_buffer = part_states
132            .iter()
133            .map(|item| item.etag.clone())
134            .collect::<Vec<_>>()
135            .join(",");
136        tracing::debug!("Finish multipart upload task body: {:?}", body_buffer);
137        headers.insert(
138            "Content-Length",
139            body_buffer.len().to_string().parse().unwrap(),
140        );
141        if let Some(ref metadata) = metadata {
142            for (k, v) in metadata {
143                headers.insert(
144                    format!("X-Ufile-Meta-{k}").parse::<HeaderName>().unwrap(),
145                    v.parse().unwrap(),
146                );
147            }
148        }
149        let resp = self
150            .client
151            .get_client()
152            .post(url)
153            .headers(headers)
154            .body(body_buffer)
155            .send()
156            .await?;
157        tracing::info!("Finish multipart upload task: {:?}", resp);
158        if resp.status().is_success() {
159            let response_headers = resp.headers();
160            let response_headers = response_headers
161                .iter()
162                .map(|(k, v)| {
163                    (
164                        k.to_string(),
165                        String::from_utf8_lossy(v.as_bytes()).to_string(),
166                    )
167                })
168                .collect::<HashMap<String, String>>();
169            let mut response_body: FinishUploadResponse = resp.json().await?;
170            if let Some(etag) = response_headers.get("etag") {
171                response_body.etag = etag.to_string();
172            }
173            response_body.headers.extend(response_headers);
174            return Ok(response_body);
175        }
176        let base_response: BaseResponse = resp.json().await?;
177        tracing::error!("Finish multipart upload task failed: {:?}", base_response);
178        Err(Error::msg("Failed to finish multipart upload task."))
179    }
180}
ufile_rus3/api/multipart_finish.rs

ufile_rus3/api/
multipart_finish.rs
