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