rusty_cat/
http_breakpoint.rs

1//! Breakpoint upload/download protocol plugins.
2//!
3//! - The executor handles scheduling, chunk I/O, retries, progress, and state.
4//! - Protocol plugins handle business-specific request/response semantics.
5
6use crate::error::{InnerErrorCode, MeowError};
7use crate::transfer_task::TransferTask;
8use async_trait::async_trait;
9
10pub use crate::download_trait::{BreakpointDownload, DownloadHeadCtx, DownloadRangeGetCtx};
11pub use crate::upload_trait::{BreakpointUpload, UploadChunkCtx, UploadPrepareCtx};
12use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
13use reqwest::multipart;
14use reqwest::Method;
15
16#[derive(Debug, Clone, Default)]
17pub struct UploadResumeInfo {
18    /// File ID returned by server when upload is already complete.
19    pub completed_file_id: Option<String>,
20    /// Suggested next byte offset (`nextByte`) from server.
21    ///
22    /// Range: `>= 0`.
23    pub next_byte: Option<u64>,
24    /// Opaque, provider-defined identifier of the in-flight multipart/resumable
25    /// upload session (for example an Aliyun OSS multipart `UploadId`).
26    ///
27    /// This is **not** a credential and is safe to persist. Persisting it lets
28    /// callers abort an orphaned session out-of-band after a crash, so that
29    /// uncommitted parts/blocks stop accruing storage cost. Providers that have
30    /// no separate session id (such as Azure Block Blob, whose resume state is
31    /// the uncommitted block list keyed by the blob URL) leave this `None`.
32    ///
33    /// Note: the bundled executor only consumes [`UploadResumeInfo::completed_file_id`]
34    /// and [`UploadResumeInfo::next_byte`]; it does not forward this id to the
35    /// progress pipeline. To persist it, hold the protocol instance and read its
36    /// accessor (for OSS, [`crate::aliyun_oss_direct::AliOssDirectUpload::current_upload_id`]),
37    /// or consume this value from a custom executor.
38    pub provider_upload_id: Option<String>,
39}
40
41/// Upload HTTP request body payload.
42#[derive(Debug)]
43pub enum UploadBody {
44    Multipart(multipart::Form),
45    Binary(Vec<u8>),
46}
47
48/// Upload request description returned by protocol plugin.
49#[derive(Debug)]
50pub struct UploadRequest {
51    /// HTTP method for upload call.
52    pub method: Method,
53    /// Target request URL.
54    pub url: String,
55    /// Request headers.
56    pub headers: HeaderMap,
57    /// Body payload.
58    pub body: UploadBody,
59}
60
61impl UploadRequest {
62    /// Creates an upload request using URL/method/headers from task.
63    ///
64    /// # Examples
65    ///
66    /// ```no_run
67    /// use rusty_cat::api::{TransferTask, UploadBody, UploadRequest};
68    ///
69    /// fn make_request(task: &TransferTask, bytes: Vec<u8>) {
70    ///     let req = UploadRequest::from_task(task, UploadBody::Binary(bytes));
71    ///     let _ = req;
72    /// }
73    /// ```
74    pub fn from_task(task: &TransferTask, body: UploadBody) -> Self {
75        Self {
76            method: task.method(),
77            url: task.url().to_string(),
78            headers: task.headers().clone(),
79            body,
80        }
81    }
82}
83
84/// Parses default upload response JSON payload into [`UploadResumeInfo`].
85fn parse_default_upload_response(body: &str) -> Result<UploadResumeInfo, MeowError> {
86    if body.trim().is_empty() {
87        crate::meow_flow_log!(
88            "upload_protocol",
89            "empty upload response body, fallback default"
90        );
91        return Ok(UploadResumeInfo::default());
92    }
93    let v: DefaultUploadResp = serde_json::from_str(body).map_err(|e| {
94        crate::meow_flow_log!(
95            "upload_protocol",
96            "upload response parse failed: body_len={} err={}",
97            body.len(),
98            e
99        );
100        MeowError::from_code(
101            InnerErrorCode::ResponseParseError,
102            format!("upload response json: {e}, body: {body}"),
103        )
104    })?;
105    crate::meow_flow_log!(
106        "upload_protocol",
107        "upload response parsed: file_id_present={} next_byte={:?}",
108        v.file_id.is_some(),
109        v.next_byte
110    );
111    Ok(UploadResumeInfo {
112        completed_file_id: v.file_id,
113        next_byte: v.next_byte.map(|n| if n < 0 { 0u64 } else { n as u64 }),
114        // The default multipart protocol carries no provider-side session id.
115        provider_upload_id: None,
116    })
117}
118
119/// Sends upload request and returns response body as string.
120async fn send_upload_request(
121    client: &reqwest::Client,
122    req: UploadRequest,
123) -> Result<String, MeowError> {
124    let mut builder = client.request(req.method, req.url).headers(req.headers);
125    builder = match req.body {
126        UploadBody::Multipart(form) => builder.multipart(form),
127        UploadBody::Binary(bytes) => builder.body(bytes),
128    };
129    let resp = builder.send().await.map_err(map_reqwest)?;
130    let status = resp.status();
131    let body = resp.text().await.map_err(map_reqwest)?;
132    if !status.is_success() {
133        return Err(MeowError::from_code(
134            InnerErrorCode::ResponseStatusError,
135            format!("upload HTTP {status}: {body}"),
136        ));
137    }
138    Ok(body)
139}
140
141/// Maps `reqwest` errors into SDK errors.
142fn map_reqwest(e: reqwest::Error) -> MeowError {
143    MeowError::from_source(InnerErrorCode::HttpError, e.to_string(), e)
144}
145
146#[derive(Debug, Clone)]
147pub struct DefaultStyleUpload {
148    /// Optional business category sent in multipart form.
149    pub category: String,
150}
151
152impl Default for DefaultStyleUpload {
153    fn default() -> Self {
154        Self {
155            category: String::new(),
156        }
157    }
158}
159
160/// Multipart field key: file md5/signature.
161const KEY_FILE_MD5: &str = "fileMd5";
162/// Multipart field key: file name.
163const KEY_FILE_NAME: &str = "fileName";
164/// Multipart field key: business category.
165const KEY_CATEGORY: &str = "category";
166/// Multipart field key: total file size.
167const KEY_TOTAL_SIZE: &str = "totalSize";
168/// Multipart field key: current chunk offset.
169const KEY_OFFSET: &str = "offset";
170/// Multipart field key: current chunk byte length.
171const KEY_PART_SIZE: &str = "partSize";
172/// Multipart field key: file binary part.
173const KEY_FILE: &str = "file";
174/// Multipart part file name used for chunk payload.
175const KEY_UPLOAD_CHUNK_DATA: &str = "upload_chunk_data";
176
177#[derive(serde::Deserialize)]
178struct DefaultUploadResp {
179    #[serde(rename = "fileId")]
180    file_id: Option<String>,
181    #[serde(rename = "nextByte")]
182    next_byte: Option<i64>,
183}
184
185#[async_trait]
186impl BreakpointUpload for DefaultStyleUpload {
187    /// Sends prepare request for default multipart upload protocol.
188    async fn prepare(&self, ctx: UploadPrepareCtx<'_>) -> Result<UploadResumeInfo, MeowError> {
189        let form = multipart::Form::new()
190            .text(KEY_FILE_MD5, ctx.task.file_sign().to_string())
191            .text(KEY_FILE_NAME, ctx.task.file_name().to_string())
192            .text(KEY_CATEGORY, self.category.clone())
193            .text(KEY_TOTAL_SIZE, ctx.task.total_size().to_string());
194        let req = UploadRequest::from_task(ctx.task, UploadBody::Multipart(form));
195        let body = send_upload_request(ctx.client, req).await?;
196        parse_default_upload_response(&body)
197    }
198
199    /// Sends one upload chunk for default multipart upload protocol.
200    ///
201    /// The chunk payload is forwarded to `reqwest` as a stream built from the
202    /// shared [`bytes::Bytes`] handle, so no per-chunk `Vec::to_vec` copy is
203    /// required. Retries clone the same `Bytes` (refcount only) instead of
204    /// re-allocating the chunk buffer.
205    async fn upload_chunk(&self, ctx: UploadChunkCtx<'_>) -> Result<UploadResumeInfo, MeowError> {
206        let chunk_len = ctx.chunk.len();
207        let body = reqwest::Body::from(ctx.chunk.clone());
208        let part = multipart::Part::stream_with_length(body, chunk_len as u64)
209            .file_name(KEY_UPLOAD_CHUNK_DATA)
210            .mime_str("application/octet-stream")
211            .map_err(|e| MeowError::from_code(InnerErrorCode::HttpError, e.to_string()))?;
212
213        let form = multipart::Form::new()
214            .part(KEY_FILE, part)
215            .text(KEY_FILE_MD5, ctx.task.file_sign().to_string())
216            .text(KEY_FILE_NAME, ctx.task.file_name().to_string())
217            .text(KEY_CATEGORY, self.category.clone())
218            .text(KEY_OFFSET, ctx.offset.to_string())
219            .text(KEY_PART_SIZE, chunk_len.to_string())
220            .text(KEY_TOTAL_SIZE, ctx.task.total_size().to_string());
221        let req = UploadRequest::from_task(ctx.task, UploadBody::Multipart(form));
222        let body = send_upload_request(ctx.client, req).await?;
223        parse_default_upload_response(&body)
224    }
225}
226
227/// HTTP behavior config for breakpoint range download.
228///
229/// This is usually provided by caller in task config. If missing, enqueue logic
230/// fills it from [`crate::meow_config::MeowConfig`].
231#[derive(Debug, Clone, PartialEq, Eq)]
232pub struct BreakpointDownloadHttpConfig {
233    /// `Accept` header used by range GET chunk requests.
234    ///
235    /// Typical value: `application/octet-stream`.
236    pub range_accept: String,
237}
238
239impl Default for BreakpointDownloadHttpConfig {
240    fn default() -> Self {
241        Self {
242            range_accept: DEFAULT_RANGE_ACCEPT.to_string(),
243        }
244    }
245}
246
247pub(crate) const DEFAULT_RANGE_ACCEPT: &str = "application/octet-stream";
248
249/// Inserts header pair into map when both name/value are valid.
250pub(crate) fn insert_header(map: &mut HeaderMap, name: &str, value: &str) {
251    if let (Ok(n), Ok(v)) = (
252        HeaderName::from_bytes(name.as_bytes()),
253        HeaderValue::from_str(value),
254    ) {
255        map.insert(n, v);
256    }
257}
258
259/// Default range download protocol.
260///
261/// It sets `Range` and `Accept` headers and reads total size from
262/// `Content-Length`.
263#[derive(Debug, Clone, Default)]
264pub struct StandardRangeDownload;
265
266impl BreakpointDownload for StandardRangeDownload {}
rusty_cat/http_breakpoint.rs

rusty_cat/
http_breakpoint.rs
