1use std::fmt::Debug;
19use std::sync::Arc;
20
21use http::Response;
22use http::StatusCode;
23use http::Uri;
24use log::debug;
25use reqsign::TencentCosConfig;
26use reqsign::TencentCosCredentialLoader;
27use reqsign::TencentCosSigner;
28
29use super::core::*;
30use super::delete::CosDeleter;
31use super::error::parse_error;
32use super::lister::CosLister;
33use super::lister::CosListers;
34use super::lister::CosObjectVersionsLister;
35use super::writer::CosWriter;
36use super::writer::CosWriters;
37use crate::raw::oio::PageLister;
38use crate::raw::*;
39use crate::services::CosConfig;
40use crate::*;
41
42impl Configurator for CosConfig {
43 type Builder = CosBuilder;
44
45 #[allow(deprecated)]
46 fn into_builder(self) -> Self::Builder {
47 CosBuilder {
48 config: self,
49
50 http_client: None,
51 }
52 }
53}
54
55#[doc = include_str!("docs.md")]
57#[derive(Default, Clone)]
58pub struct CosBuilder {
59 config: CosConfig,
60
61 #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
62 http_client: Option<HttpClient>,
63}
64
65impl Debug for CosBuilder {
66 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67 f.debug_struct("CosBuilder")
68 .field("config", &self.config)
69 .finish()
70 }
71}
72
73impl CosBuilder {
74 pub fn root(mut self, root: &str) -> Self {
78 self.config.root = if root.is_empty() {
79 None
80 } else {
81 Some(root.to_string())
82 };
83
84 self
85 }
86
87 pub fn endpoint(mut self, endpoint: &str) -> Self {
95 if !endpoint.is_empty() {
96 self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string());
97 }
98
99 self
100 }
101
102 pub fn secret_id(mut self, secret_id: &str) -> Self {
106 if !secret_id.is_empty() {
107 self.config.secret_id = Some(secret_id.to_string());
108 }
109
110 self
111 }
112
113 pub fn secret_key(mut self, secret_key: &str) -> Self {
117 if !secret_key.is_empty() {
118 self.config.secret_key = Some(secret_key.to_string());
119 }
120
121 self
122 }
123
124 pub fn bucket(mut self, bucket: &str) -> Self {
127 if !bucket.is_empty() {
128 self.config.bucket = Some(bucket.to_string());
129 }
130
131 self
132 }
133
134 pub fn enable_versioning(mut self, enabled: bool) -> Self {
136 self.config.enable_versioning = enabled;
137
138 self
139 }
140
141 pub fn disable_config_load(mut self) -> Self {
148 self.config.disable_config_load = true;
149 self
150 }
151
152 #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
159 #[allow(deprecated)]
160 pub fn http_client(mut self, client: HttpClient) -> Self {
161 self.http_client = Some(client);
162 self
163 }
164}
165
166impl Builder for CosBuilder {
167 const SCHEME: Scheme = Scheme::Cos;
168 type Config = CosConfig;
169
170 fn build(self) -> Result<impl Access> {
171 debug!("backend build started: {:?}", &self);
172
173 let root = normalize_root(&self.config.root.unwrap_or_default());
174 debug!("backend use root {root}");
175
176 let bucket = match &self.config.bucket {
177 Some(bucket) => Ok(bucket.to_string()),
178 None => Err(
179 Error::new(ErrorKind::ConfigInvalid, "The bucket is misconfigured")
180 .with_context("service", Scheme::Cos),
181 ),
182 }?;
183 debug!("backend use bucket {}", &bucket);
184
185 let uri = match &self.config.endpoint {
186 Some(endpoint) => endpoint.parse::<Uri>().map_err(|err| {
187 Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
188 .with_context("service", Scheme::Cos)
189 .with_context("endpoint", endpoint)
190 .set_source(err)
191 }),
192 None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
193 .with_context("service", Scheme::Cos)),
194 }?;
195
196 let scheme = match uri.scheme_str() {
197 Some(scheme) => scheme.to_string(),
198 None => "https".to_string(),
199 };
200
201 let endpoint = uri.host().unwrap().replace(&format!("//{bucket}."), "//");
203 debug!("backend use endpoint {}", &endpoint);
204
205 let mut cfg = TencentCosConfig::default();
206 if !self.config.disable_config_load {
207 cfg = cfg.from_env();
208 }
209
210 if let Some(v) = self.config.secret_id {
211 cfg.secret_id = Some(v);
212 }
213 if let Some(v) = self.config.secret_key {
214 cfg.secret_key = Some(v);
215 }
216
217 let cred_loader = TencentCosCredentialLoader::new(GLOBAL_REQWEST_CLIENT.clone(), cfg);
218
219 let signer = TencentCosSigner::new();
220
221 Ok(CosBackend {
222 core: Arc::new(CosCore {
223 info: {
224 let am = AccessorInfo::default();
225 am.set_scheme(Scheme::Cos)
226 .set_root(&root)
227 .set_name(&bucket)
228 .set_native_capability(Capability {
229 stat: true,
230 stat_with_if_match: true,
231 stat_with_if_none_match: true,
232 stat_with_version: self.config.enable_versioning,
233
234 read: true,
235
236 read_with_if_match: true,
237 read_with_if_none_match: true,
238 read_with_version: self.config.enable_versioning,
239
240 write: true,
241 write_can_empty: true,
242 write_can_append: true,
243 write_can_multi: true,
244 write_with_content_type: true,
245 write_with_cache_control: true,
246 write_with_content_disposition: true,
247 write_with_if_not_exists: !self.config.enable_versioning,
249 write_multi_min_size: Some(1024 * 1024),
253 write_multi_max_size: if cfg!(target_pointer_width = "64") {
257 Some(5 * 1024 * 1024 * 1024)
258 } else {
259 Some(usize::MAX)
260 },
261 write_with_user_metadata: true,
262
263 delete: true,
264 delete_with_version: self.config.enable_versioning,
265 copy: true,
266
267 list: true,
268 list_with_recursive: true,
269 list_with_versions: self.config.enable_versioning,
270 list_with_deleted: self.config.enable_versioning,
271
272 presign: true,
273 presign_stat: true,
274 presign_read: true,
275 presign_write: true,
276
277 shared: true,
278
279 ..Default::default()
280 });
281
282 #[allow(deprecated)]
284 if let Some(client) = self.http_client {
285 am.update_http_client(|_| client);
286 }
287
288 am.into()
289 },
290 bucket: bucket.clone(),
291 root,
292 endpoint: format!("{}://{}.{}", &scheme, &bucket, &endpoint),
293 signer,
294 loader: cred_loader,
295 }),
296 })
297 }
298}
299
300#[derive(Debug, Clone)]
302pub struct CosBackend {
303 core: Arc<CosCore>,
304}
305
306impl Access for CosBackend {
307 type Reader = HttpBody;
308 type Writer = CosWriters;
309 type Lister = CosListers;
310 type Deleter = oio::OneShotDeleter<CosDeleter>;
311
312 fn info(&self) -> Arc<AccessorInfo> {
313 self.core.info.clone()
314 }
315
316 async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
317 let resp = self.core.cos_head_object(path, &args).await?;
318
319 let status = resp.status();
320
321 match status {
322 StatusCode::OK => {
323 let headers = resp.headers();
324 let mut meta = parse_into_metadata(path, headers)?;
325
326 let user_meta = parse_prefixed_headers(headers, "x-cos-meta-");
327 if !user_meta.is_empty() {
328 meta = meta.with_user_metadata(user_meta);
329 }
330
331 if let Some(v) = parse_header_to_str(headers, constants::X_COS_VERSION_ID)? {
332 if v != "null" {
333 meta.set_version(v);
334 }
335 }
336
337 Ok(RpStat::new(meta))
338 }
339 _ => Err(parse_error(resp)),
340 }
341 }
342
343 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
344 let resp = self.core.cos_get_object(path, args.range(), &args).await?;
345
346 let status = resp.status();
347
348 match status {
349 StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
350 Ok((RpRead::default(), resp.into_body()))
351 }
352 _ => {
353 let (part, mut body) = resp.into_parts();
354 let buf = body.to_buffer().await?;
355 Err(parse_error(Response::from_parts(part, buf)))
356 }
357 }
358 }
359
360 async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
361 let writer = CosWriter::new(self.core.clone(), path, args.clone());
362
363 let w = if args.append() {
364 CosWriters::Two(oio::AppendWriter::new(writer))
365 } else {
366 CosWriters::One(oio::MultipartWriter::new(
367 self.core.info.clone(),
368 writer,
369 args.concurrent(),
370 ))
371 };
372
373 Ok((RpWrite::default(), w))
374 }
375
376 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
377 Ok((
378 RpDelete::default(),
379 oio::OneShotDeleter::new(CosDeleter::new(self.core.clone())),
380 ))
381 }
382
383 async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
384 let l = if args.versions() || args.deleted() {
385 TwoWays::Two(PageLister::new(CosObjectVersionsLister::new(
386 self.core.clone(),
387 path,
388 args,
389 )))
390 } else {
391 TwoWays::One(PageLister::new(CosLister::new(
392 self.core.clone(),
393 path,
394 args.recursive(),
395 args.limit(),
396 )))
397 };
398
399 Ok((RpList::default(), l))
400 }
401
402 async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
403 let resp = self.core.cos_copy_object(from, to).await?;
404
405 let status = resp.status();
406
407 match status {
408 StatusCode::OK => Ok(RpCopy::default()),
409 _ => Err(parse_error(resp)),
410 }
411 }
412
413 async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
414 let req = match args.operation() {
415 PresignOperation::Stat(v) => self.core.cos_head_object_request(path, v),
416 PresignOperation::Read(v) => {
417 self.core
418 .cos_get_object_request(path, BytesRange::default(), v)
419 }
420 PresignOperation::Write(v) => {
421 self.core
422 .cos_put_object_request(path, None, v, Buffer::new())
423 }
424 PresignOperation::Delete(_) => Err(Error::new(
425 ErrorKind::Unsupported,
426 "operation is not supported",
427 )),
428 };
429 let mut req = req?;
430 self.core.sign_query(&mut req, args.expire()).await?;
431
432 let (parts, _) = req.into_parts();
434
435 Ok(RpPresign::new(PresignedRequest::new(
436 parts.method,
437 parts.uri,
438 parts.headers,
439 )))
440 }
441}