1use std::collections::HashMap;
5use thiserror::Error;
6
7#[derive(Debug, Error, Clone, PartialEq, Eq)]
13pub enum CloudError {
14 #[error("invalid URL: {0}")]
16 InvalidUrl(String),
17
18 #[error("unsupported scheme: {0}")]
20 UnsupportedScheme(String),
21
22 #[error("missing credentials")]
24 MissingCredentials,
25
26 #[error("invalid credentials: {0}")]
28 InvalidCredentials(String),
29
30 #[error("presign error: {0}")]
32 PresignError(String),
33
34 #[error("range out of bounds: [{start}, {end}) vs size {size}")]
36 RangeOutOfBounds {
37 start: u64,
39 end: u64,
41 size: u64,
43 },
44
45 #[error("HTTP error: status {status}, url: {url}")]
47 HttpError {
48 status: u16,
50 url: String,
52 },
53
54 #[error("HTTP transport error: {0}")]
56 TransportError(String),
57
58 #[error("exhausted all {attempts} retry attempts")]
60 RetryExhausted {
61 attempts: u32,
63 },
64
65 #[error("unsupported credentials: {0}")]
67 UnsupportedCredentials(String),
68
69 #[error("missing or malformed header: {0}")]
71 MissingHeader(String),
72}
73
74#[derive(Debug, Clone, PartialEq, Eq)]
80pub enum CloudScheme {
81 S3,
83 Gs,
85 Az,
87 Http,
89 Https,
91}
92
93#[derive(Debug, Clone, PartialEq, Eq)]
95pub struct ObjectUrl {
96 pub scheme: CloudScheme,
98 pub bucket: String,
100 pub key: String,
102 pub region: Option<String>,
104 pub endpoint: Option<String>,
106}
107
108impl ObjectUrl {
109 pub fn parse(url: &str) -> Result<Self, CloudError> {
119 let (scheme_str, rest) = url
120 .split_once("://")
121 .ok_or_else(|| CloudError::InvalidUrl(format!("no scheme separator in '{url}'")))?;
122
123 let scheme = match scheme_str.to_ascii_lowercase().as_str() {
124 "s3" => CloudScheme::S3,
125 "gs" => CloudScheme::Gs,
126 "az" | "abfs" => CloudScheme::Az,
127 "http" => CloudScheme::Http,
128 "https" => CloudScheme::Https,
129 other => return Err(CloudError::UnsupportedScheme(other.to_owned())),
130 };
131
132 match &scheme {
133 CloudScheme::Http | CloudScheme::Https => {
134 let (host, path) = if let Some(idx) = rest.find('/') {
136 (&rest[..idx], &rest[idx + 1..])
137 } else {
138 (rest, "")
139 };
140 if host.is_empty() {
141 return Err(CloudError::InvalidUrl(format!("no host in '{url}'")));
142 }
143 Ok(ObjectUrl {
144 scheme,
145 bucket: host.to_owned(),
146 key: path.to_owned(),
147 region: None,
148 endpoint: None,
149 })
150 }
151 _ => {
152 let (bucket, key) = if let Some(idx) = rest.find('/') {
154 (&rest[..idx], &rest[idx + 1..])
155 } else {
156 (rest, "")
157 };
158 if bucket.is_empty() {
159 return Err(CloudError::InvalidUrl(format!("no bucket in '{url}'")));
160 }
161 Ok(ObjectUrl {
162 scheme,
163 bucket: bucket.to_owned(),
164 key: key.to_owned(),
165 region: None,
166 endpoint: None,
167 })
168 }
169 }
170 }
171
172 pub fn to_https_url(&self, endpoint_override: Option<&str>) -> String {
179 if let Some(ep) = endpoint_override {
180 let base = ep.trim_end_matches('/');
181 return format!("{base}/{}/{}", self.bucket, self.key);
182 }
183 match &self.scheme {
184 CloudScheme::S3 => {
185 let region = self.region.as_deref().unwrap_or("us-east-1");
186 format!(
187 "https://{}.s3.{}.amazonaws.com/{}",
188 self.bucket, region, self.key
189 )
190 }
191 CloudScheme::Gs => {
192 format!(
193 "https://storage.googleapis.com/{}/{}",
194 self.bucket, self.key
195 )
196 }
197 CloudScheme::Az => {
198 format!("https://{}.blob.core.windows.net/{}", self.bucket, self.key)
201 }
202 CloudScheme::Http => {
203 format!("https://{}/{}", self.bucket, self.key)
204 }
205 CloudScheme::Https => {
206 format!("https://{}/{}", self.bucket, self.key)
207 }
208 }
209 }
210
211 pub fn signing_host(&self) -> String {
213 match &self.scheme {
214 CloudScheme::S3 => {
215 let region = self.region.as_deref().unwrap_or("us-east-1");
216 format!("{}.s3.{}.amazonaws.com", self.bucket, region)
217 }
218 CloudScheme::Gs => "storage.googleapis.com".to_owned(),
219 CloudScheme::Az => {
220 format!("{}.blob.core.windows.net", self.bucket)
221 }
222 CloudScheme::Http | CloudScheme::Https => self.bucket.clone(),
223 }
224 }
225
226 pub fn signing_path(&self) -> String {
228 let key = if self.key.starts_with('/') {
229 self.key.clone()
230 } else {
231 format!("/{}", self.key)
232 };
233 match &self.scheme {
234 CloudScheme::Gs | CloudScheme::Az => {
235 format!("/{}{}", self.bucket, key)
236 }
237 _ => key,
238 }
239 }
240}
241
242#[derive(Debug, Clone, PartialEq, Eq)]
248pub struct ByteRangeRequest {
249 pub url: ObjectUrl,
251 pub range: std::ops::Range<u64>,
253}
254
255impl ByteRangeRequest {
256 pub fn new(url: ObjectUrl, start: u64, end: u64) -> Self {
258 ByteRangeRequest {
259 url,
260 range: start..end,
261 }
262 }
263
264 pub fn to_http_range_header(&self) -> String {
266 let end_inclusive = self.range.end.saturating_sub(1);
267 format!("bytes={}-{}", self.range.start, end_inclusive)
268 }
269
270 pub fn length(&self) -> u64 {
272 self.range.end.saturating_sub(self.range.start)
273 }
274}
275
276#[derive(Debug, Clone, PartialEq, Eq)]
282pub struct ObjectMetadata {
283 pub url: ObjectUrl,
285 pub size: u64,
287 pub content_type: Option<String>,
289 pub etag: Option<String>,
291 pub last_modified: Option<u64>,
293 pub user_metadata: HashMap<String, String>,
295}
296
297#[derive(Debug, Clone, PartialEq, Eq)]
303pub enum CloudCredentials {
304 Anonymous,
306 AccessKey {
308 access_key_id: String,
310 secret_access_key: String,
312 session_token: Option<String>,
314 },
315 ServiceAccountFile {
317 path: String,
319 },
320 AzureSharedKey {
322 account_name: String,
324 account_key: String,
326 },
327 SasToken {
329 token: String,
331 },
332 Bearer {
334 token: String,
336 },
337}
338
339#[derive(Debug, Clone, PartialEq, Eq)]
345pub enum HttpMethod {
346 Get,
348 Put,
350 Delete,
352 Head,
354}
355
356impl HttpMethod {
357 fn as_str(&self) -> &'static str {
358 match self {
359 HttpMethod::Get => "GET",
360 HttpMethod::Put => "PUT",
361 HttpMethod::Delete => "DELETE",
362 HttpMethod::Head => "HEAD",
363 }
364 }
365}
366
367#[derive(Debug, Clone, PartialEq, Eq)]
369pub struct PresignedUrlConfig {
370 pub expires_in_secs: u64,
372 pub method: HttpMethod,
374 pub content_type: Option<String>,
376}
377
378impl PresignedUrlConfig {
379 pub fn get(expires_in_secs: u64) -> Self {
381 PresignedUrlConfig {
382 expires_in_secs,
383 method: HttpMethod::Get,
384 content_type: None,
385 }
386 }
387
388 pub fn put(expires_in_secs: u64, content_type: impl Into<String>) -> Self {
390 PresignedUrlConfig {
391 expires_in_secs,
392 method: HttpMethod::Put,
393 content_type: Some(content_type.into()),
394 }
395 }
396}
397
398const K: [u32; 64] = [
404 0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5,
405 0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174,
406 0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da,
407 0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967,
408 0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85,
409 0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070,
410 0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3,
411 0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2,
412];
413
414const H0: [u32; 8] = [
416 0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab, 0x5be0cd19,
417];
418
419pub fn sha256(data: &[u8]) -> [u8; 32] {
421 let mut h = H0;
422
423 let bit_len = (data.len() as u64).wrapping_mul(8);
425 let mut msg: Vec<u8> = data.to_vec();
426 msg.push(0x80);
427 while msg.len() % 64 != 56 {
428 msg.push(0x00);
429 }
430 msg.extend_from_slice(&bit_len.to_be_bytes());
431
432 for block in msg.chunks_exact(64) {
434 let mut w = [0u32; 64];
435 for (i, chunk) in block.chunks_exact(4).enumerate().take(16) {
436 w[i] = u32::from_be_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
437 }
438 for i in 16..64 {
439 let s0 = w[i - 15].rotate_right(7) ^ w[i - 15].rotate_right(18) ^ (w[i - 15] >> 3);
440 let s1 = w[i - 2].rotate_right(17) ^ w[i - 2].rotate_right(19) ^ (w[i - 2] >> 10);
441 w[i] = w[i - 16]
442 .wrapping_add(s0)
443 .wrapping_add(w[i - 7])
444 .wrapping_add(s1);
445 }
446
447 let [mut a, mut b, mut c, mut d, mut e, mut f, mut g, mut hh] = h;
448
449 for i in 0..64 {
450 let s1 = e.rotate_right(6) ^ e.rotate_right(11) ^ e.rotate_right(25);
451 let ch = (e & f) ^ ((!e) & g);
452 let temp1 = hh
453 .wrapping_add(s1)
454 .wrapping_add(ch)
455 .wrapping_add(K[i])
456 .wrapping_add(w[i]);
457 let s0 = a.rotate_right(2) ^ a.rotate_right(13) ^ a.rotate_right(22);
458 let maj = (a & b) ^ (a & c) ^ (b & c);
459 let temp2 = s0.wrapping_add(maj);
460
461 hh = g;
462 g = f;
463 f = e;
464 e = d.wrapping_add(temp1);
465 d = c;
466 c = b;
467 b = a;
468 a = temp1.wrapping_add(temp2);
469 }
470
471 h[0] = h[0].wrapping_add(a);
472 h[1] = h[1].wrapping_add(b);
473 h[2] = h[2].wrapping_add(c);
474 h[3] = h[3].wrapping_add(d);
475 h[4] = h[4].wrapping_add(e);
476 h[5] = h[5].wrapping_add(f);
477 h[6] = h[6].wrapping_add(g);
478 h[7] = h[7].wrapping_add(hh);
479 }
480
481 let mut out = [0u8; 32];
482 for (i, &word) in h.iter().enumerate() {
483 out[i * 4..(i + 1) * 4].copy_from_slice(&word.to_be_bytes());
484 }
485 out
486}
487
488pub fn hmac_sha256(key: &[u8], data: &[u8]) -> [u8; 32] {
490 const BLOCK_SIZE: usize = 64;
491
492 let mut k = [0u8; BLOCK_SIZE];
494 if key.len() > BLOCK_SIZE {
495 let hashed = sha256(key);
496 k[..32].copy_from_slice(&hashed);
497 } else {
498 k[..key.len()].copy_from_slice(key);
499 }
500
501 let mut ipad = [0u8; BLOCK_SIZE];
502 let mut opad = [0u8; BLOCK_SIZE];
503 for i in 0..BLOCK_SIZE {
504 ipad[i] = k[i] ^ 0x36;
505 opad[i] = k[i] ^ 0x5c;
506 }
507
508 let mut inner = ipad.to_vec();
509 inner.extend_from_slice(data);
510 let inner_hash = sha256(&inner);
511
512 let mut outer = opad.to_vec();
513 outer.extend_from_slice(&inner_hash);
514 sha256(&outer)
515}
516
517pub fn hmac_sha256_hex(key: &[u8], data: &[u8]) -> String {
519 hex_encode(&hmac_sha256(key, data))
520}
521
522pub fn hex_encode(bytes: &[u8]) -> String {
524 bytes
525 .iter()
526 .fold(String::with_capacity(bytes.len() * 2), |mut s, b| {
527 use std::fmt::Write;
528 let _ = write!(s, "{b:02x}");
529 s
530 })
531}
532
533pub struct PresignedUrlGenerator {
541 pub credentials: CloudCredentials,
543 pub region: String,
545}
546
547impl PresignedUrlGenerator {
548 pub fn new(credentials: CloudCredentials, region: impl Into<String>) -> Self {
550 PresignedUrlGenerator {
551 credentials,
552 region: region.into(),
553 }
554 }
555
556 fn format_date(ts: u64) -> String {
558 let days = ts / 86_400;
560 let (year, month, day) = days_to_ymd(days);
561 format!("{year:04}{month:02}{day:02}")
562 }
563
564 fn format_datetime(ts: u64) -> String {
566 let date = Self::format_date(ts);
567 let rem = ts % 86_400;
568 let h = rem / 3600;
569 let m = (rem % 3600) / 60;
570 let s = rem % 60;
571 format!("{date}T{h:02}{m:02}{s:02}Z")
572 }
573
574 fn derive_signing_key(secret: &str, date: &str, region: &str, service: &str) -> [u8; 32] {
576 let k_date = hmac_sha256(format!("AWS4{secret}").as_bytes(), date.as_bytes());
577 let k_region = hmac_sha256(&k_date, region.as_bytes());
578 let k_service = hmac_sha256(&k_region, service.as_bytes());
579 hmac_sha256(&k_service, b"aws4_request")
580 }
581
582 fn uri_encode(s: &str, encode_slash: bool) -> String {
584 let mut out = String::with_capacity(s.len());
585 for byte in s.bytes() {
586 match byte {
587 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
588 out.push(byte as char);
589 }
590 b'/' if !encode_slash => out.push('/'),
591 other => {
592 use std::fmt::Write;
593 let _ = write!(out, "%{other:02X}");
594 }
595 }
596 }
597 out
598 }
599
600 pub fn canonical_query_string(&self, params: &[(String, String)]) -> String {
602 let mut sorted: Vec<(String, String)> = params
603 .iter()
604 .map(|(k, v)| (Self::uri_encode(k, true), Self::uri_encode(v, true)))
605 .collect();
606 sorted.sort_by(|(a, _), (b, _)| a.cmp(b));
607 sorted
608 .iter()
609 .map(|(k, v)| format!("{k}={v}"))
610 .collect::<Vec<_>>()
611 .join("&")
612 }
613
614 pub fn generate_s3(
616 &self,
617 url: &ObjectUrl,
618 config: &PresignedUrlConfig,
619 timestamp_unix: u64,
620 ) -> Result<String, CloudError> {
621 let (access_key_id, secret_access_key) = match &self.credentials {
622 CloudCredentials::AccessKey {
623 access_key_id,
624 secret_access_key,
625 ..
626 } => (access_key_id.as_str(), secret_access_key.as_str()),
627 _ => return Err(CloudError::MissingCredentials),
628 };
629
630 let service = "s3";
631 let date = Self::format_date(timestamp_unix);
632 let datetime = Self::format_datetime(timestamp_unix);
633
634 let credential = format!(
635 "{access_key_id}/{date}/{}/{service}/aws4_request",
636 self.region
637 );
638
639 let host = url.signing_host();
640 let path = Self::uri_encode(&url.key, false);
641 let canonical_path = format!("/{path}");
642
643 let mut query_params: Vec<(String, String)> = vec![
644 ("X-Amz-Algorithm".to_owned(), "AWS4-HMAC-SHA256".to_owned()),
645 ("X-Amz-Credential".to_owned(), credential.clone()),
646 ("X-Amz-Date".to_owned(), datetime.clone()),
647 (
648 "X-Amz-Expires".to_owned(),
649 config.expires_in_secs.to_string(),
650 ),
651 ("X-Amz-SignedHeaders".to_owned(), "host".to_owned()),
652 ];
653
654 if let CloudCredentials::AccessKey {
655 session_token: Some(tok),
656 ..
657 } = &self.credentials
658 {
659 query_params.push(("X-Amz-Security-Token".to_owned(), tok.clone()));
660 }
661
662 let canonical_query = self.canonical_query_string(&query_params);
663
664 let canonical_headers = format!("host:{host}\n");
665 let signed_headers = "host";
666
667 let payload_hash = "UNSIGNED-PAYLOAD";
669
670 let canonical_request = format!(
671 "{method}\n{path}\n{query}\n{headers}\n{signed}\n{payload}",
672 method = config.method.as_str(),
673 path = canonical_path,
674 query = canonical_query,
675 headers = canonical_headers,
676 signed = signed_headers,
677 payload = payload_hash,
678 );
679
680 let scope = format!("{date}/{}/{service}/aws4_request", self.region);
681 let string_to_sign = format!(
682 "AWS4-HMAC-SHA256\n{datetime}\n{scope}\n{hash}",
683 hash = hex_encode(&sha256(canonical_request.as_bytes())),
684 );
685
686 let signing_key = Self::derive_signing_key(secret_access_key, &date, &self.region, service);
687 let signature = hmac_sha256_hex(&signing_key, string_to_sign.as_bytes());
688
689 let mut final_params = query_params;
690 final_params.push(("X-Amz-Signature".to_owned(), signature));
691 let final_query = self.canonical_query_string(&final_params);
692
693 let base_url = format!("https://{host}{canonical_path}");
694 Ok(format!("{base_url}?{final_query}"))
695 }
696
697 pub fn generate_gcs(
699 &self,
700 url: &ObjectUrl,
701 config: &PresignedUrlConfig,
702 timestamp_unix: u64,
703 ) -> Result<String, CloudError> {
704 let (access_key_id, secret_access_key) = match &self.credentials {
705 CloudCredentials::AccessKey {
706 access_key_id,
707 secret_access_key,
708 ..
709 } => (access_key_id.as_str(), secret_access_key.as_str()),
710 CloudCredentials::ServiceAccountFile { path } => {
711 return Err(CloudError::PresignError(format!(
714 "service account file signing requires JSON parsing (path: {path})"
715 )));
716 }
717 _ => return Err(CloudError::MissingCredentials),
718 };
719
720 let service = "storage";
721 let date = Self::format_date(timestamp_unix);
722 let datetime = Self::format_datetime(timestamp_unix);
723 let host = "storage.googleapis.com";
724 let canonical_path = format!("/{}/{}", url.bucket, Self::uri_encode(&url.key, false));
725
726 let credential = format!(
727 "{access_key_id}/{date}/{}/{service}/goog4_request",
728 self.region
729 );
730
731 let query_params: Vec<(String, String)> = vec![
732 (
733 "X-Goog-Algorithm".to_owned(),
734 "GOOG4-HMAC-SHA256".to_owned(),
735 ),
736 ("X-Goog-Credential".to_owned(), credential.clone()),
737 ("X-Goog-Date".to_owned(), datetime.clone()),
738 (
739 "X-Goog-Expires".to_owned(),
740 config.expires_in_secs.to_string(),
741 ),
742 ("X-Goog-SignedHeaders".to_owned(), "host".to_owned()),
743 ];
744
745 let canonical_query = self.canonical_query_string(&query_params);
746 let canonical_headers = format!("host:{host}\n");
747 let signed_headers = "host";
748 let payload_hash = "UNSIGNED-PAYLOAD";
749
750 let canonical_request = format!(
751 "{method}\n{path}\n{query}\n{headers}\n{signed}\n{payload}",
752 method = config.method.as_str(),
753 path = canonical_path,
754 query = canonical_query,
755 headers = canonical_headers,
756 signed = signed_headers,
757 payload = payload_hash,
758 );
759
760 let scope = format!("{date}/{}/{service}/goog4_request", self.region);
761 let string_to_sign = format!(
762 "GOOG4-HMAC-SHA256\n{datetime}\n{scope}\n{hash}",
763 hash = hex_encode(&sha256(canonical_request.as_bytes())),
764 );
765
766 let signing_key = Self::derive_signing_key(secret_access_key, &date, &self.region, service);
767 let signature = hmac_sha256_hex(&signing_key, string_to_sign.as_bytes());
768
769 let mut final_params = query_params;
770 final_params.push(("X-Goog-Signature".to_owned(), signature));
771 let final_query = self.canonical_query_string(&final_params);
772
773 Ok(format!("https://{host}{canonical_path}?{final_query}"))
774 }
775}
776
777fn days_to_ymd(days: u64) -> (u32, u32, u32) {
783 let z = days as i64 + 719_468;
786 let era: i64 = if z >= 0 { z } else { z - 146_096 } / 146_097;
787 let doe = (z - era * 146_097) as u64;
788 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
789 let y = yoe as i64 + era * 400;
790 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
791 let mp = (5 * doy + 2) / 153;
792 let d = doy - (153 * mp + 2) / 5 + 1;
793 let m = if mp < 10 { mp + 3 } else { mp - 9 };
794 let y_adj = if m <= 2 { y + 1 } else { y };
795 (y_adj as u32, m as u32, d as u32)
796}
797
798#[derive(Debug, Clone, PartialEq, Eq)]
804pub struct CompletedPart {
805 pub part_number: u16,
807 pub etag: String,
809 pub size: u64,
811}
812
813#[derive(Debug, Clone, PartialEq, Eq)]
815pub struct MultipartUploadState {
816 pub upload_id: String,
818 pub url: ObjectUrl,
820 pub parts: Vec<CompletedPart>,
822 pub part_size: u64,
824}
825
826impl MultipartUploadState {
827 pub fn new(upload_id: impl Into<String>, url: ObjectUrl, part_size: u64) -> Self {
829 MultipartUploadState {
830 upload_id: upload_id.into(),
831 url,
832 parts: Vec::new(),
833 part_size,
834 }
835 }
836
837 pub fn add_part(&mut self, part_number: u16, etag: impl Into<String>, size: u64) {
839 self.parts.push(CompletedPart {
840 part_number,
841 etag: etag.into(),
842 size,
843 });
844 }
845
846 pub fn total_size(&self) -> u64 {
848 self.parts.iter().map(|p| p.size).sum()
849 }
850
851 pub fn part_count(&self) -> usize {
853 self.parts.len()
854 }
855
856 pub fn to_xml(&self) -> String {
860 let mut sorted = self.parts.clone();
861 sorted.sort_by_key(|p| p.part_number);
862
863 let mut xml = String::from("<CompleteMultipartUpload>\n");
864 for part in &sorted {
865 xml.push_str(" <Part>\n");
866 xml.push_str(&format!(
867 " <PartNumber>{}</PartNumber>\n",
868 part.part_number
869 ));
870 xml.push_str(&format!(" <ETag>{}</ETag>\n", part.etag));
871 xml.push_str(" </Part>\n");
872 }
873 xml.push_str("</CompleteMultipartUpload>");
874 xml
875 }
876}
877
878pub struct CloudRangeCoalescer {
885 pub max_gap_bytes: u64,
887 pub max_request_size: u64,
889 pub min_request_size: u64,
891}
892
893impl Default for CloudRangeCoalescer {
894 fn default() -> Self {
895 Self::new()
896 }
897}
898
899impl CloudRangeCoalescer {
900 pub fn new() -> Self {
905 CloudRangeCoalescer {
906 max_gap_bytes: 512 * 1024,
907 max_request_size: 8 * 1024 * 1024,
908 min_request_size: 64 * 1024,
909 }
910 }
911
912 pub fn coalesce(&self, mut ranges: Vec<ByteRangeRequest>) -> Vec<ByteRangeRequest> {
919 if ranges.is_empty() {
920 return ranges;
921 }
922
923 ranges.sort_by_key(|r| r.range.start);
925
926 let url = ranges[0].url.clone();
927 let mut coalesced: Vec<ByteRangeRequest> = Vec::new();
928 let mut current_start = ranges[0].range.start;
929 let mut current_end = ranges[0].range.end;
930
931 for req in ranges.into_iter().skip(1) {
932 let gap = req.range.start.saturating_sub(current_end);
933 let new_end = req.range.end.max(current_end);
934 let new_size = new_end - current_start;
935
936 if gap <= self.max_gap_bytes && new_size <= self.max_request_size {
937 current_end = new_end;
939 } else {
940 coalesced.push(ByteRangeRequest::new(
941 url.clone(),
942 current_start,
943 current_end,
944 ));
945 current_start = req.range.start;
946 current_end = req.range.end;
947 }
948 }
949 coalesced.push(ByteRangeRequest::new(url, current_start, current_end));
950 coalesced
951 }
952
953 pub fn slice_response<'a>(
962 coalesced_data: &'a [u8],
963 coalesced_start: u64,
964 sub_range: &std::ops::Range<u64>,
965 ) -> &'a [u8] {
966 let offset = (sub_range.start - coalesced_start) as usize;
967 let len = (sub_range.end - sub_range.start) as usize;
968 &coalesced_data[offset..offset + len]
969 }
970}