zng_task/http/
curl.rs

1use std::{fmt, time::Duration};
2
3use crate::{
4    http::{Error, HttpClient, Metrics, Request, Response},
5    io::{BufReader, ReadLimited},
6};
7use futures_lite::{AsyncBufReadExt as _, AsyncReadExt, AsyncWriteExt as _, io::Cursor};
8use http::Uri;
9use once_cell::sync::Lazy;
10use zng_unit::{ByteLength, ByteUnits as _};
11use zng_var::{Var, const_var, var};
12
13use super::uri::Scheme;
14
15/// Basic [`HttpClient`] implementation that uses the `curl` command line utility.
16#[derive(Default)]
17pub struct CurlProcessClient {}
18impl HttpClient for CurlProcessClient {
19    fn send(&'static self, request: Request) -> std::pin::Pin<Box<dyn Future<Output = Result<Response, Error>> + Send>> {
20        Box::pin(run(request))
21    }
22
23    fn is_cache_manager(&self) -> bool {
24        false
25    }
26}
27
28async fn run(request: Request) -> Result<Response, Error> {
29    let not_http = match request.uri.scheme() {
30        Some(s) => s != &Scheme::HTTP && s != &Scheme::HTTPS,
31        None => true,
32    };
33    if not_http {
34        return Err(Box::new(NotHttpUriError));
35    }
36
37    let mut curl = crate::process::Command::new(&*CURL);
38
39    curl.stdin(std::process::Stdio::piped())
40        .stdout(std::process::Stdio::piped())
41        .stderr(std::process::Stdio::piped());
42    curl.arg("--include"); // print header
43
44    curl.arg("--http1.1");
45
46    curl.arg("-N"); // no internal buffering for pipes
47
48    curl.arg("-X").arg(request.method.as_str());
49
50    #[cfg(feature = "http_compression")]
51    if request.auto_decompress && !request.headers.contains_key(http::header::ACCEPT_ENCODING) {
52        curl.arg("-H").arg("accept-encoding").arg("zstd, br, gzip");
53    }
54    for (name, value) in request.headers {
55        if let Some(name) = name
56            && let Ok(value) = value.to_str()
57        {
58            curl.arg("-H").arg(format!("{name}: {value}"));
59        }
60    }
61
62    let connect_timeout = request.timeout.min(request.connect_timeout);
63    if connect_timeout < Duration::MAX {
64        curl.arg("--connect-timeout").arg(request.connect_timeout.as_secs().to_string());
65    }
66    if request.timeout < Duration::MAX {
67        curl.arg("--max-time").arg(request.timeout.as_secs().to_string());
68    }
69    if request.low_speed_timeout.0 < Duration::MAX && request.low_speed_timeout.1 > 0.bytes() {
70        curl.arg("-y")
71            .arg(request.low_speed_timeout.0.as_secs().to_string())
72            .arg("-Y")
73            .arg(request.low_speed_timeout.1.bytes().to_string());
74    }
75
76    if request.redirect_limit > 0 {
77        curl.arg("-L").arg("--max-redirs").arg(request.redirect_limit.to_string());
78    }
79    let rate_limit = request.max_upload_speed.min(request.max_download_speed);
80    if rate_limit < ByteLength::MAX {
81        curl.arg("--limit-rate").arg(format!("{}K", rate_limit.kibis()));
82    }
83
84    if !request.body.is_empty() {
85        curl.arg("--data-binary").arg("@-");
86    }
87
88    curl.arg(request.uri.to_string());
89
90    let mut curl = curl.spawn()?;
91
92    let mut stdin = curl.stdin.take().unwrap();
93    let mut stdout = BufReader::new(curl.stdout.take().unwrap());
94    let stderr = curl.stderr.take().unwrap();
95
96    if !request.body.is_empty() {
97        stdin.write_all(&request.body[..]).await?;
98        stdin.flush().await?;
99    }
100    stdin.close().await?;
101    drop(stdin);
102
103    let metrics = if request.metrics {
104        let m = var(Metrics::zero());
105        read_metrics(m.clone(), stderr);
106        m.read_only()
107    } else {
108        const_var(Metrics::zero())
109    };
110
111    let mut response_bytes = Vec::with_capacity(1024);
112    let mut buffer = [0u8; 1024];
113    let mut effective_uri = request.uri;
114    loop {
115        let bytes_read = stdout.read(&mut buffer).await?;
116        if bytes_read == 0 && response_bytes.is_empty() {
117            Err(Box::new(UnexpectedPartialError))?;
118        }
119
120        response_bytes.extend_from_slice(&buffer[..bytes_read]);
121
122        let mut response_headers = [httparse::EMPTY_HEADER; 64];
123        let mut response = httparse::Response::new(&mut response_headers);
124
125        match response.parse(&response_bytes)? {
126            httparse::Status::Complete(header_length) => {
127                let code = http::StatusCode::from_u16(response.code.unwrap_or(502))?;
128
129                if code.is_redirection()
130                    && let Some(l) = response.headers.iter().find(|h| h.name.eq_ignore_ascii_case("Location"))
131                    && let Ok(l) = std::str::from_utf8(l.value)
132                    && let Ok(l) = l.parse::<Uri>()
133                {
134                    effective_uri = l;
135
136                    let content_length = response
137                        .headers
138                        .iter()
139                        .find(|h| h.name.eq_ignore_ascii_case("Content-Length"))
140                        .and_then(|h| std::str::from_utf8(h.value).ok())
141                        .and_then(|v| v.parse::<usize>().ok())
142                        .unwrap_or(0);
143                    let redirect_rsp_len = header_length + content_length;
144                    if response_bytes.len() > redirect_rsp_len {
145                        response_bytes.drain(..redirect_rsp_len);
146                    }
147
148                    continue;
149                }
150
151                let initial_body_chunk = &response_bytes[header_length..];
152
153                return run_response(
154                    response,
155                    effective_uri,
156                    #[cfg(feature = "http_compression")]
157                    request.auto_decompress,
158                    request.require_length,
159                    request.max_length,
160                    metrics,
161                    initial_body_chunk,
162                    stdout,
163                );
164            }
165            httparse::Status::Partial => {
166                continue;
167            }
168        }
169    }
170}
171fn read_metrics(metrics: Var<Metrics>, stderr: crate::process::ChildStderr) {
172    let mut stderr = BufReader::new(stderr);
173    let mut progress_bytes = Vec::with_capacity(92);
174    let mut run = async move || -> std::io::Result<()> {
175        loop {
176            progress_bytes.clear();
177            let len = stderr.read_until(b'\r', &mut progress_bytes).await?;
178            if len == 0 {
179                break;
180            }
181
182            let progress = str::from_utf8(&progress_bytes).map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
183            if !progress.trim_start().chars().next().unwrap_or('\0').is_ascii_digit() {
184                continue;
185            }
186            // https://everything.curl.dev/cmdline/progressmeter.html#progress-meter-legend
187            let mut iter = progress.split_whitespace();
188            let _pct = iter.next();
189            let _total = iter.next();
190            let pct_down: u8 = iter.next().unwrap_or("100").parse().unwrap_or(100);
191            let down = parse_curl_bytes(iter.next().unwrap_or("0"));
192            let response_total = (down.0 as f64 * 100.0 / pct_down as f64).bytes();
193            let pct_up: u8 = iter.next().unwrap_or("100").parse().unwrap_or(100);
194            let up = parse_curl_bytes(iter.next().unwrap_or("0"));
195            let request_total = (up.0 as f64 * 100.0 / pct_up as f64).bytes();
196            let down_speed = parse_curl_bytes(iter.next().unwrap_or("0"));
197            let up_speed = parse_curl_bytes(iter.next().unwrap_or("0"));
198            let _total_time = iter.next();
199            let time_current = parse_curl_duration(iter.next().unwrap_or("HH:MM:SS"));
200
201            metrics.set(Metrics {
202                read_progress: (down, response_total),
203                read_speed: down_speed,
204                write_progress: (up, request_total),
205                write_speed: up_speed,
206                total_time: time_current,
207            });
208        }
209
210        Ok(())
211    };
212    crate::spawn(async move {
213        let _ = run().await;
214    });
215}
216fn parse_curl_bytes(s: &str) -> ByteLength {
217    // https://everything.curl.dev/cmdline/progressmeter.html#units
218    let (s, scale) = if let Some(s) = s.strip_suffix("K") {
219        (s, 2u64.pow(10))
220    } else if let Some(s) = s.strip_suffix("M") {
221        (s, 2u64.pow(20))
222    } else if let Some(s) = s.strip_prefix("G") {
223        (s, 2u64.pow(30))
224    } else if let Some(s) = s.strip_prefix("T") {
225        (s, 2u64.pow(40))
226    } else if let Some(s) = s.strip_prefix("P") {
227        (s, 2u64.pow(50))
228    } else {
229        (s, 1)
230    };
231    let l: u64 = s.parse().unwrap_or(0);
232    ByteLength::from_byte(l * scale)
233}
234fn parse_curl_duration(s: &str) -> Duration {
235    // HH:MM:SS
236    let mut iter = s.split(':');
237    let h: usize = iter.next().unwrap_or("0").parse().unwrap_or(0);
238    let m: u8 = iter.next().unwrap_or("0").parse().unwrap_or(0);
239    let s: u8 = iter.next().unwrap_or("0").parse().unwrap_or(0);
240    Duration::from_hours(h as _) + Duration::from_mins(m as _) + Duration::from_secs(s as _)
241}
242
243fn run_response(
244    response: httparse::Response<'_, '_>,
245    effective_uri: Uri,
246    #[cfg(feature = "http_compression")] auto_decompress: bool,
247    require_length: bool,
248    max_length: ByteLength,
249    metrics: Var<Metrics>,
250    initial_body_chunk: &[u8],
251    reader: BufReader<crate::process::ChildStdout>,
252) -> Result<Response, Error> {
253    let reader = Cursor::new(initial_body_chunk.to_owned()).chain(reader);
254    let code = http::StatusCode::from_u16(response.code.unwrap_or(502))?;
255
256    let mut header = http::header::HeaderMap::new();
257    for r in response.headers {
258        if r.name.is_empty() {
259            continue;
260        }
261        header.append(
262            http::HeaderName::from_bytes(r.name.as_bytes())?,
263            http::HeaderValue::from_bytes(r.value)?,
264        );
265    }
266    if require_length {
267        if let Some(l) = header.get(http::header::CONTENT_LENGTH)
268            && let Ok(l) = l.to_str()
269            && let Ok(l) = l.parse::<u64>()
270        {
271            if l < max_length.bytes() {
272                return Err(Box::new(ContentLengthExceedsMaxError));
273            }
274        } else {
275            return Err(Box::new(ContentLengthRequiredError));
276        }
277    }
278
279    let reader = ReadLimited::new_default_err(reader, max_length);
280
281    macro_rules! respond {
282        ($read:expr) => {
283            return Ok(Response::from_read(code, header, effective_uri, metrics, Box::new($read)))
284        };
285    }
286
287    #[cfg(feature = "http_compression")]
288    if auto_decompress && let Some(enc) = header.get(http::header::CONTENT_ENCODING) {
289        if enc == "zstd" {
290            respond!(async_compression::futures::bufread::ZstdDecoder::new(reader))
291        } else if enc == "br" {
292            respond!(async_compression::futures::bufread::BrotliDecoder::new(reader))
293        } else if enc == "gzip" {
294            respond!(async_compression::futures::bufread::GzipDecoder::new(reader))
295        }
296    }
297    respond!(reader)
298}
299
300static CURL: Lazy<String> = Lazy::new(|| std::env::var("ZNG_CURL").unwrap_or_else(|_| "curl".to_owned()));
301
302#[derive(Debug)]
303struct NotHttpUriError;
304impl fmt::Display for NotHttpUriError {
305    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
306        write!(f, "uri is not HTTP or HTTPS")
307    }
308}
309impl std::error::Error for NotHttpUriError {}
310
311#[derive(Debug)]
312struct ContentLengthRequiredError;
313impl fmt::Display for ContentLengthRequiredError {
314    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
315        write!(f, "response content length is required")
316    }
317}
318impl std::error::Error for ContentLengthRequiredError {}
319
320#[derive(Debug)]
321struct ContentLengthExceedsMaxError;
322impl fmt::Display for ContentLengthExceedsMaxError {
323    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
324        write!(f, "response content length is exceeds maximum")
325    }
326}
327impl std::error::Error for ContentLengthExceedsMaxError {}
328
329#[derive(Debug)]
330struct UnexpectedPartialError;
331impl fmt::Display for UnexpectedPartialError {
332    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
333        write!(f, "unexpected partial response from curl")
334    }
335}
336impl std::error::Error for UnexpectedPartialError {}
zng_task/http/curl.rs

zng_task/http/
curl.rs
