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 async_trait::async_trait;
7use crate::error::{InnerErrorCode, MeowError};
8use crate::transfer_task::TransferTask;
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}
25
26/// Upload HTTP request body payload.
27#[derive(Debug)]
28pub enum UploadBody {
29    Multipart(multipart::Form),
30    Binary(Vec<u8>),
31}
32
33/// Upload request description returned by protocol plugin.
34#[derive(Debug)]
35pub struct UploadRequest {
36    /// HTTP method for upload call.
37    pub method: Method,
38    /// Target request URL.
39    pub url: String,
40    /// Request headers.
41    pub headers: HeaderMap,
42    /// Body payload.
43    pub body: UploadBody,
44}
45
46impl UploadRequest {
47    /// Creates an upload request using URL/method/headers from task.
48    ///
49    /// # Examples
50    ///
51    /// ```no_run
52    /// use rusty_cat::api::{TransferTask, UploadBody, UploadRequest};
53    ///
54    /// fn make_request(task: &TransferTask, bytes: Vec<u8>) {
55    ///     let req = UploadRequest::from_task(task, UploadBody::Binary(bytes));
56    ///     let _ = req;
57    /// }
58    /// ```
59    pub fn from_task(task: &TransferTask, body: UploadBody) -> Self {
60        Self {
61            method: task.method(),
62            url: task.url().to_string(),
63            headers: task.headers().clone(),
64            body,
65        }
66    }
67}
68
69/// Parses default upload response JSON payload into [`UploadResumeInfo`].
70fn parse_default_upload_response(body: &str) -> Result<UploadResumeInfo, MeowError> {
71    if body.trim().is_empty() {
72        crate::meow_flow_log!(
73            "upload_protocol",
74            "empty upload response body, fallback default"
75        );
76        return Ok(UploadResumeInfo::default());
77    }
78    let v: DefaultUploadResp = serde_json::from_str(body).map_err(|e| {
79        crate::meow_flow_log!(
80            "upload_protocol",
81            "upload response parse failed: body_len={} err={}",
82            body.len(),
83            e
84        );
85        MeowError::from_code(
86            InnerErrorCode::ResponseParseError,
87            format!("upload response json: {e}, body: {body}"),
88        )
89    })?;
90    crate::meow_flow_log!(
91        "upload_protocol",
92        "upload response parsed: file_id_present={} next_byte={:?}",
93        v.file_id.is_some(),
94        v.next_byte
95    );
96    Ok(UploadResumeInfo {
97        completed_file_id: v.file_id,
98        next_byte: v.next_byte.map(|n| if n < 0 { 0u64 } else { n as u64 }),
99    })
100}
101
102/// Sends upload request and returns response body as string.
103async fn send_upload_request(
104    client: &reqwest::Client,
105    req: UploadRequest,
106) -> Result<String, MeowError> {
107    let mut builder = client.request(req.method, req.url).headers(req.headers);
108    builder = match req.body {
109        UploadBody::Multipart(form) => builder.multipart(form),
110        UploadBody::Binary(bytes) => builder.body(bytes),
111    };
112    let resp = builder.send().await.map_err(map_reqwest)?;
113    let status = resp.status();
114    let body = resp.text().await.map_err(map_reqwest)?;
115    if !status.is_success() {
116        return Err(MeowError::from_code(
117            InnerErrorCode::ResponseStatusError,
118            format!("upload HTTP {status}: {body}"),
119        ));
120    }
121    Ok(body)
122}
123
124/// Maps `reqwest` errors into SDK errors.
125fn map_reqwest(e: reqwest::Error) -> MeowError {
126    MeowError::from_source(InnerErrorCode::HttpError, e.to_string(), e)
127}
128
129#[derive(Debug, Clone)]
130pub struct DefaultStyleUpload {
131    /// Optional business category sent in multipart form.
132    pub category: String,
133}
134
135impl Default for DefaultStyleUpload {
136    fn default() -> Self {
137        Self {
138            category: String::new(),
139        }
140    }
141}
142
143/// Multipart field key: file md5/signature.
144const KEY_FILE_MD5: &str = "fileMd5";
145/// Multipart field key: file name.
146const KEY_FILE_NAME: &str = "fileName";
147/// Multipart field key: business category.
148const KEY_CATEGORY: &str = "category";
149/// Multipart field key: total file size.
150const KEY_TOTAL_SIZE: &str = "totalSize";
151/// Multipart field key: current chunk offset.
152const KEY_OFFSET: &str = "offset";
153/// Multipart field key: current chunk byte length.
154const KEY_PART_SIZE: &str = "partSize";
155/// Multipart field key: file binary part.
156const KEY_FILE: &str = "file";
157/// Multipart part file name used for chunk payload.
158const KEY_UPLOAD_CHUNK_DATA: &str = "upload_chunk_data";
159
160#[derive(serde::Deserialize)]
161struct DefaultUploadResp {
162    #[serde(rename = "fileId")]
163    file_id: Option<String>,
164    #[serde(rename = "nextByte")]
165    next_byte: Option<i64>,
166}
167
168#[async_trait]
169impl BreakpointUpload for DefaultStyleUpload {
170    /// Sends prepare request for default multipart upload protocol.
171    async fn prepare(&self, ctx: UploadPrepareCtx<'_>) -> Result<UploadResumeInfo, MeowError> {
172        let form = multipart::Form::new()
173            .text(KEY_FILE_MD5, ctx.task.file_sign().to_string())
174            .text(KEY_FILE_NAME, ctx.task.file_name().to_string())
175            .text(KEY_CATEGORY, self.category.clone())
176            .text(KEY_TOTAL_SIZE, ctx.task.total_size().to_string());
177        let req = UploadRequest::from_task(ctx.task, UploadBody::Multipart(form));
178        let body = send_upload_request(ctx.client, req).await?;
179        parse_default_upload_response(&body)
180    }
181
182    /// Sends one upload chunk for default multipart upload protocol.
183    async fn upload_chunk(&self, ctx: UploadChunkCtx<'_>) -> Result<UploadResumeInfo, MeowError> {
184        let part = multipart::Part::bytes(ctx.chunk.to_vec())
185            .file_name(KEY_UPLOAD_CHUNK_DATA)
186            .mime_str("application/octet-stream")
187            .map_err(|e| MeowError::from_code(InnerErrorCode::HttpError, e.to_string()))?;
188
189        let form = multipart::Form::new()
190            .part(KEY_FILE, part)
191            .text(KEY_FILE_MD5, ctx.task.file_sign().to_string())
192            .text(KEY_FILE_NAME, ctx.task.file_name().to_string())
193            .text(KEY_CATEGORY, self.category.clone())
194            .text(KEY_OFFSET, ctx.offset.to_string())
195            .text(KEY_PART_SIZE, ctx.chunk.len().to_string())
196            .text(KEY_TOTAL_SIZE, ctx.task.total_size().to_string());
197        let req = UploadRequest::from_task(ctx.task, UploadBody::Multipart(form));
198        let body = send_upload_request(ctx.client, req).await?;
199        parse_default_upload_response(&body)
200    }
201}
202
203/// HTTP behavior config for breakpoint range download.
204///
205/// This is usually provided by caller in task config. If missing, enqueue logic
206/// fills it from [`crate::meow_config::MeowConfig`].
207#[derive(Debug, Clone, PartialEq, Eq)]
208pub struct BreakpointDownloadHttpConfig {
209    /// `Accept` header used by range GET chunk requests.
210    ///
211    /// Typical value: `application/octet-stream`.
212    pub range_accept: String,
213}
214
215impl Default for BreakpointDownloadHttpConfig {
216    fn default() -> Self {
217        Self {
218            range_accept: DEFAULT_RANGE_ACCEPT.to_string(),
219        }
220    }
221}
222
223pub(crate) const DEFAULT_RANGE_ACCEPT: &str = "application/octet-stream";
224
225/// Inserts header pair into map when both name/value are valid.
226pub(crate) fn insert_header(map: &mut HeaderMap, name: &str, value: &str) {
227    if let (Ok(n), Ok(v)) = (
228        HeaderName::from_bytes(name.as_bytes()),
229        HeaderValue::from_str(value),
230    ) {
231        map.insert(n, v);
232    }
233}
234
235/// Default range download protocol.
236///
237/// It sets `Range` and `Accept` headers and reads total size from
238/// `Content-Length`.
239#[derive(Debug, Clone, Default)]
240pub struct StandardRangeDownload;
241
242impl BreakpointDownload for StandardRangeDownload {}
rusty_cat/http_breakpoint.rs

rusty_cat/
http_breakpoint.rs
