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

zng_task/http/
curl.rs
