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_has_cache_control: true,
233 stat_has_content_length: true,
234 stat_has_content_type: true,
235 stat_has_content_encoding: true,
236 stat_has_content_range: true,
237 stat_with_version: self.config.enable_versioning,
238 stat_has_etag: true,
239 stat_has_content_md5: true,
240 stat_has_last_modified: true,
241 stat_has_content_disposition: true,
242 stat_has_version: true,
243 stat_has_user_metadata: true,
244
245 read: true,
246
247 read_with_if_match: true,
248 read_with_if_none_match: true,
249 read_with_version: self.config.enable_versioning,
250
251 write: true,
252 write_can_empty: true,
253 write_can_append: true,
254 write_can_multi: true,
255 write_with_content_type: true,
256 write_with_cache_control: true,
257 write_with_content_disposition: true,
258 write_with_if_not_exists: !self.config.enable_versioning,
260 write_multi_min_size: Some(1024 * 1024),
264 write_multi_max_size: if cfg!(target_pointer_width = "64") {
268 Some(5 * 1024 * 1024 * 1024)
269 } else {
270 Some(usize::MAX)
271 },
272 write_with_user_metadata: true,
273
274 delete: true,
275 delete_with_version: self.config.enable_versioning,
276 copy: true,
277
278 list: true,
279 list_with_recursive: true,
280 list_with_versions: self.config.enable_versioning,
281 list_with_deleted: self.config.enable_versioning,
282 list_has_content_length: true,
283
284 presign: true,
285 presign_stat: true,
286 presign_read: true,
287 presign_write: true,
288
289 shared: true,
290
291 ..Default::default()
292 });
293
294 #[allow(deprecated)]
296 if let Some(client) = self.http_client {
297 am.update_http_client(|_| client);
298 }
299
300 am.into()
301 },
302 bucket: bucket.clone(),
303 root,
304 endpoint: format!("{}://{}.{}", &scheme, &bucket, &endpoint),
305 signer,
306 loader: cred_loader,
307 }),
308 })
309 }
310}
311
312#[derive(Debug, Clone)]
314pub struct CosBackend {
315 core: Arc<CosCore>,
316}
317
318impl Access for CosBackend {
319 type Reader = HttpBody;
320 type Writer = CosWriters;
321 type Lister = CosListers;
322 type Deleter = oio::OneShotDeleter<CosDeleter>;
323
324 fn info(&self) -> Arc<AccessorInfo> {
325 self.core.info.clone()
326 }
327
328 async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
329 let resp = self.core.cos_head_object(path, &args).await?;
330
331 let status = resp.status();
332
333 match status {
334 StatusCode::OK => {
335 let headers = resp.headers();
336 let mut meta = parse_into_metadata(path, headers)?;
337
338 let user_meta = parse_prefixed_headers(headers, "x-cos-meta-");
339 if !user_meta.is_empty() {
340 meta = meta.with_user_metadata(user_meta);
341 }
342
343 if let Some(v) = parse_header_to_str(headers, constants::X_COS_VERSION_ID)? {
344 if v != "null" {
345 meta.set_version(v);
346 }
347 }
348
349 Ok(RpStat::new(meta))
350 }
351 _ => Err(parse_error(resp)),
352 }
353 }
354
355 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
356 let resp = self.core.cos_get_object(path, args.range(), &args).await?;
357
358 let status = resp.status();
359
360 match status {
361 StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
362 Ok((RpRead::default(), resp.into_body()))
363 }
364 _ => {
365 let (part, mut body) = resp.into_parts();
366 let buf = body.to_buffer().await?;
367 Err(parse_error(Response::from_parts(part, buf)))
368 }
369 }
370 }
371
372 async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
373 let writer = CosWriter::new(self.core.clone(), path, args.clone());
374
375 let w = if args.append() {
376 CosWriters::Two(oio::AppendWriter::new(writer))
377 } else {
378 CosWriters::One(oio::MultipartWriter::new(
379 self.core.info.clone(),
380 writer,
381 args.concurrent(),
382 ))
383 };
384
385 Ok((RpWrite::default(), w))
386 }
387
388 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
389 Ok((
390 RpDelete::default(),
391 oio::OneShotDeleter::new(CosDeleter::new(self.core.clone())),
392 ))
393 }
394
395 async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
396 let l = if args.versions() || args.deleted() {
397 TwoWays::Two(PageLister::new(CosObjectVersionsLister::new(
398 self.core.clone(),
399 path,
400 args,
401 )))
402 } else {
403 TwoWays::One(PageLister::new(CosLister::new(
404 self.core.clone(),
405 path,
406 args.recursive(),
407 args.limit(),
408 )))
409 };
410
411 Ok((RpList::default(), l))
412 }
413
414 async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
415 let resp = self.core.cos_copy_object(from, to).await?;
416
417 let status = resp.status();
418
419 match status {
420 StatusCode::OK => Ok(RpCopy::default()),
421 _ => Err(parse_error(resp)),
422 }
423 }
424
425 async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
426 let req = match args.operation() {
427 PresignOperation::Stat(v) => self.core.cos_head_object_request(path, v),
428 PresignOperation::Read(v) => {
429 self.core
430 .cos_get_object_request(path, BytesRange::default(), v)
431 }
432 PresignOperation::Write(v) => {
433 self.core
434 .cos_put_object_request(path, None, v, Buffer::new())
435 }
436 PresignOperation::Delete(_) => Err(Error::new(
437 ErrorKind::Unsupported,
438 "operation is not supported",
439 )),
440 };
441 let mut req = req?;
442 self.core.sign_query(&mut req, args.expire()).await?;
443
444 let (parts, _) = req.into_parts();
446
447 Ok(RpPresign::new(PresignedRequest::new(
448 parts.method,
449 parts.uri,
450 parts.headers,
451 )))
452 }
453}