1#[cfg(test)]
58mod tests;
59
60use std::io::{Read, Write};
61#[cfg(not(target_arch = "wasm32"))]
62use std::net::TcpStream;
63#[cfg(not(target_arch = "wasm32"))]
64use std::time::Duration;
65
66#[cfg(any(feature = "http-client", feature = "http2"))]
67use std::sync::Arc;
68
69fn form_encode(s: &str) -> String {
73 let mut out = String::with_capacity(s.len());
74 for b in s.bytes() {
75 match b {
76 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
77 out.push(b as char)
78 }
79 _ => out.push_str(&format!("%{:02X}", b)),
80 }
81 }
82 out
83}
84
85fn form_urlencode(pairs: &[(&str, &str)]) -> String {
87 pairs
88 .iter()
89 .map(|(k, v)| format!("{}={}", form_encode(k), form_encode(v)))
90 .collect::<Vec<_>>()
91 .join("&")
92}
93
94#[derive(Debug)]
98pub struct HttpClientError(pub String);
99
100impl std::fmt::Display for HttpClientError {
101 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102 f.write_str(&self.0)
103 }
104}
105
106impl std::error::Error for HttpClientError {}
107
108struct ParsedUrl {
111 scheme: String,
112 host: String,
113 port: u16,
114 path_and_query: String,
115}
116
117impl ParsedUrl {
118 fn host_header(&self) -> String {
122 let default_port = if self.scheme == "https" { 443 } else { 80 };
123 let host = if self.host.contains(':') && !self.host.starts_with('[') { format!("[{}]", self.host) } else { self.host.clone() };
124 if self.port == default_port { host } else { format!("{host}:{}", self.port) }
125 }
126
127 fn parse(url: &str) -> Result<Self, HttpClientError> {
128 let rest = if let Some(r) = url.strip_prefix("https://") {
130 ("https", r)
131 } else if let Some(r) = url.strip_prefix("http://") {
132 ("http", r)
133 } else {
134 return Err(HttpClientError(format!(
135 "unsupported or missing URL scheme in '{url}'"
136 )));
137 };
138
139 let (scheme, authority_and_path) = rest;
140 let default_port: u16 = if scheme == "https" { 443 } else { 80 };
141
142 let (authority, path_and_query) = match authority_and_path.find('/') {
144 Some(idx) => {
145 let (a, p) = authority_and_path.split_at(idx);
146 (a, p.to_string())
147 }
148 None => (authority_and_path, "/".to_string()),
149 };
150
151 let (host, port) = if let Some(bracket_end) = authority.find(']') {
153 let host = &authority[..=bracket_end];
155 let port_part = &authority[bracket_end + 1..];
156 let port = if let Some(p) = port_part.strip_prefix(':') {
157 p.parse::<u16>().map_err(|_| {
158 HttpClientError(format!("invalid port in URL '{url}'"))
159 })?
160 } else {
161 default_port
162 };
163 (host.to_string(), port)
164 } else {
165 match authority.rfind(':') {
166 Some(idx) => {
167 let port_str = &authority[idx + 1..];
168 let port = port_str.parse::<u16>().map_err(|_| {
169 HttpClientError(format!("invalid port in URL '{url}'"))
170 })?;
171 (authority[..idx].to_string(), port)
172 }
173 None => (authority.to_string(), default_port),
174 }
175 };
176
177 if host.is_empty() {
178 return Err(HttpClientError(format!("missing host in URL '{url}'")));
179 }
180
181 Ok(ParsedUrl {
182 scheme: scheme.to_string(),
183 host,
184 port,
185 path_and_query,
186 })
187 }
188}
189
190fn resolve_url(base_url: &str, location: &str) -> String {
194 if location.starts_with("http://") || location.starts_with("https://") {
195 return location.to_string();
196 }
197 if let Ok(base) = ParsedUrl::parse(base_url) {
199 let default_port = if base.scheme == "https" { 443 } else { 80 };
200 let port_str = if base.port == default_port {
201 String::new()
202 } else {
203 format!(":{}", base.port)
204 };
205 if location.starts_with('/') {
206 return format!("{}://{}{}{}", base.scheme, base.host, port_str, location);
207 }
208 let base_path = base.path_and_query;
210 let dir = match base_path.rfind('/') {
211 Some(i) => &base_path[..=i],
212 None => "/",
213 };
214 return format!(
215 "{}://{}{}{}{}",
216 base.scheme, base.host, port_str, dir, location
217 );
218 }
219 location.to_string()
220}
221
222#[derive(Debug)]
226pub struct Response {
227 status: u16,
228 headers: Vec<(String, String)>,
229 body: Vec<u8>,
230}
231
232impl Response {
233 pub fn status(&self) -> u16 {
235 self.status
236 }
237
238 pub fn is_success(&self) -> bool {
240 (200..300).contains(&self.status)
241 }
242
243 pub fn is_redirect(&self) -> bool {
245 matches!(self.status, 301 | 302 | 303 | 307 | 308)
246 }
247
248 pub fn header(&self, name: &str) -> Option<&str> {
250 let lower = name.to_lowercase();
251 self.headers
252 .iter()
253 .find(|(k, _)| k.to_lowercase() == lower)
254 .map(|(_, v)| v.as_str())
255 }
256
257 pub fn headers(&self) -> &[(String, String)] {
261 &self.headers
262 }
263
264 pub fn bytes(&self) -> &[u8] {
266 &self.body
267 }
268
269 pub fn text(&self) -> Result<String, HttpClientError> {
271 String::from_utf8(self.body.clone())
272 .map_err(|e| HttpClientError(format!("body is not valid UTF-8: {e}")))
273 }
274
275 #[cfg(feature = "serde")]
277 pub fn json<T: serde::de::DeserializeOwned>(&self) -> Result<T, HttpClientError> {
278 serde_json::from_slice(&self.body)
279 .map_err(|e| HttpClientError(format!("JSON parse error: {e}")))
280 }
281}
282
283fn build_request_bytes(
287 method: &str,
288 path_and_query: &str,
289 host: &str,
290 headers: &[(String, String)],
291 body: &Option<Vec<u8>>,
292) -> Vec<u8> {
293 let mut out: Vec<u8> = Vec::new();
294
295 let _ = write!(
297 out,
298 "{method} {path_and_query} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\nUser-Agent: rust-web-server/{}\r\n",
299 env!("CARGO_PKG_VERSION"),
300 );
301
302 if let Some(b) = body {
304 if !b.is_empty() {
305 let _ = write!(out, "Content-Length: {}\r\n", b.len());
306 }
307 }
308
309 for (k, v) in headers {
311 let _ = write!(out, "{k}: {v}\r\n");
312 }
313
314 out.extend_from_slice(b"\r\n");
315
316 if let Some(b) = body {
317 out.extend_from_slice(b);
318 }
319
320 out
321}
322
323fn read_response(stream: &mut dyn Read, is_head: bool) -> Result<Response, HttpClientError> {
325 let mut buf: Vec<u8> = Vec::with_capacity(8192);
326 let mut tmp = [0u8; 4096];
327
328 let header_end = loop {
330 let n = stream
331 .read(&mut tmp)
332 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
333 if n == 0 {
334 if buf.is_empty() {
335 return Err(HttpClientError(
336 "server closed connection without sending a response".into(),
337 ));
338 }
339 break buf.len();
341 }
342 buf.extend_from_slice(&tmp[..n]);
343 if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
344 break pos + 4;
345 }
346 };
347
348 let header_block = std::str::from_utf8(&buf[..header_end])
350 .map_err(|_| HttpClientError("response headers are not valid UTF-8".into()))?;
351
352 let mut lines = header_block.lines();
353
354 let status_line = lines
356 .next()
357 .ok_or_else(|| HttpClientError("empty response".into()))?;
358 let status = parse_status(status_line)?;
359
360 let response_headers: Vec<(String, String)> = lines
362 .filter_map(|line| {
363 let mut parts = line.splitn(2, ':');
364 let name = parts.next()?.trim().to_string();
365 let value = parts.next()?.trim().to_string();
366 if name.is_empty() {
367 None
368 } else {
369 Some((name, value))
370 }
371 })
372 .collect();
373
374 let mut body = buf[header_end..].to_vec();
376
377 if !is_head {
378 let transfer_encoding = response_headers
380 .iter()
381 .find(|(k, _)| k.to_lowercase() == "transfer-encoding")
382 .map(|(_, v)| v.to_lowercase());
383
384 let content_length: Option<usize> = response_headers
385 .iter()
386 .find(|(k, _)| k.to_lowercase() == "content-length")
387 .and_then(|(_, v)| v.trim().parse().ok());
388
389 if transfer_encoding
390 .as_deref()
391 .map(|te| te.contains("chunked"))
392 .unwrap_or(false)
393 {
394 loop {
396 let n = stream
397 .read(&mut tmp)
398 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
399 if n == 0 {
400 break;
401 }
402 body.extend_from_slice(&tmp[..n]);
403 }
404 body = decode_chunked(&body)?;
405 } else if let Some(len) = content_length {
406 while body.len() < len {
407 let n = stream
408 .read(&mut tmp)
409 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
410 if n == 0 {
411 break;
412 }
413 body.extend_from_slice(&tmp[..n]);
414 }
415 body.truncate(len);
416 } else {
417 loop {
419 let n = stream
420 .read(&mut tmp)
421 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
422 if n == 0 {
423 break;
424 }
425 body.extend_from_slice(&tmp[..n]);
426 }
427 }
428 } else {
429 body.clear();
430 }
431
432 Ok(Response {
433 status,
434 headers: response_headers,
435 body,
436 })
437}
438
439fn parse_status(line: &str) -> Result<u16, HttpClientError> {
440 let mut parts = line.splitn(3, ' ');
442 let _version = parts
443 .next()
444 .ok_or_else(|| HttpClientError("malformed status line".into()))?;
445 let code_str = parts
446 .next()
447 .ok_or_else(|| HttpClientError("missing status code".into()))?;
448 code_str
449 .parse::<u16>()
450 .map_err(|_| HttpClientError(format!("invalid status code '{code_str}'")))
451}
452
453fn decode_chunked(data: &[u8]) -> Result<Vec<u8>, HttpClientError> {
455 let mut out = Vec::new();
456 let mut pos = 0;
457
458 while pos < data.len() {
459 let line_end = data[pos..]
461 .windows(2)
462 .position(|w| w == b"\r\n")
463 .ok_or_else(|| HttpClientError("invalid chunked encoding: missing CRLF".into()))?;
464 let size_line = std::str::from_utf8(&data[pos..pos + line_end])
465 .map_err(|_| HttpClientError("chunked size is not ASCII".into()))?
466 .trim();
467 let size_str = size_line.split(';').next().unwrap_or("").trim();
469 let chunk_size = usize::from_str_radix(size_str, 16)
470 .map_err(|_| HttpClientError(format!("invalid chunk size '{size_str}'")))?;
471 pos += line_end + 2; if chunk_size == 0 {
474 break; }
476
477 let end = pos + chunk_size;
478 if end > data.len() {
479 return Err(HttpClientError("chunked body truncated".into()));
480 }
481 out.extend_from_slice(&data[pos..end]);
482 pos = end + 2; }
484
485 Ok(out)
486}
487
488#[cfg(all(any(feature = "http-client", feature = "http2"), not(target_arch = "wasm32")))]
491fn tls_connect(
492 host: &str,
493 tcp: TcpStream,
494 roots: Option<&Arc<rustls::RootCertStore>>,
495) -> Result<rustls::StreamOwned<rustls::ClientConnection, TcpStream>, HttpClientError> {
496 use rustls::pki_types::ServerName;
497
498 let roots = match roots {
499 Some(roots) => Arc::clone(roots),
500 None => Arc::new(rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned())),
501 };
502 let config = Arc::new(crate::tls_client_builder().with_root_certificates(roots).with_no_client_auth());
503 let server_name = ServerName::try_from(host.to_string())
504 .map_err(|e| HttpClientError(format!("invalid hostname '{host}': {e}")))?;
505 let conn = rustls::ClientConnection::new(config, server_name)
506 .map_err(|e| HttpClientError(e.to_string()))?;
507 Ok(rustls::StreamOwned::new(conn, tcp))
508}
509
510#[cfg(not(target_arch = "wasm32"))]
513fn send_once(
514 method: &str,
515 parsed: &ParsedUrl,
516 headers: &[(String, String)],
517 body: &Option<Vec<u8>>,
518 timeout_ms: u64,
519 #[cfg(any(feature = "http-client", feature = "http2"))] tls: Option<&Arc<rustls::RootCertStore>>,
520) -> Result<Response, HttpClientError> {
521 let addr = format!("{}:{}", parsed.host, parsed.port);
522 let timeout = Duration::from_millis(timeout_ms);
523
524 let addresses: Vec<std::net::SocketAddr> = match addr.parse::<std::net::SocketAddr>() {
527 Ok(one) => vec![one],
528 Err(_) => {
529 use std::net::ToSocketAddrs;
530 addr.to_socket_addrs().map_err(|e| HttpClientError(format!("DNS lookup for '{addr}' failed: {e}")))?.collect()
531 }
532 };
533 let mut last_error = HttpClientError(format!("no address for '{addr}'"));
534 let mut connected = None;
535 for sock_addr in addresses {
536 match TcpStream::connect_timeout(&sock_addr, timeout) {
537 Ok(tcp) => {
538 connected = Some(tcp);
539 break;
540 }
541 Err(e) => last_error = HttpClientError(format!("connect to '{addr}' ({sock_addr}) failed: {e}")),
542 }
543 }
544 let tcp = connected.ok_or(last_error)?;
545 tcp.set_read_timeout(Some(timeout))
546 .map_err(|e| HttpClientError(e.to_string()))?;
547 tcp.set_write_timeout(Some(timeout))
548 .map_err(|e| HttpClientError(e.to_string()))?;
549
550 let request_bytes =
551 build_request_bytes(method, &parsed.path_and_query, &parsed.host_header(), headers, body);
552
553 let is_head = method.eq_ignore_ascii_case("HEAD");
554
555 #[cfg(any(feature = "http-client", feature = "http2"))]
557 if parsed.scheme == "https" {
558 let mut tls_stream = tls_connect(&parsed.host, tcp, tls)?;
559 tls_stream
560 .write_all(&request_bytes)
561 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
562 return read_response(&mut tls_stream, is_head);
563 }
564
565 let mut stream = tcp;
567 stream
568 .write_all(&request_bytes)
569 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
570 read_response(&mut stream, is_head)
571}
572
573#[cfg(target_arch = "wasm32")]
588fn send_once(
589 method: &str,
590 parsed: &ParsedUrl,
591 headers: &[(String, String)],
592 body: &Option<Vec<u8>>,
593 timeout_ms: u64,
594 _tls: Option<&()>,
595) -> Result<Response, HttpClientError> {
596 use wasip2::http::outgoing_handler;
597 use wasip2::http::types::{Fields, OutgoingBody, OutgoingRequest, RequestOptions, Scheme};
598
599 let wasi_headers = Fields::new();
600 for (name, value) in headers {
601 let _ = wasi_headers.append(name, value.as_bytes());
602 }
603
604 let request = OutgoingRequest::new(wasi_headers);
605 request
606 .set_method(&to_wasi_method(method))
607 .map_err(|_| HttpClientError(format!("invalid method '{method}'")))?;
608 let scheme = if parsed.scheme == "https" { Scheme::Https } else { Scheme::Http };
609 request
610 .set_scheme(Some(&scheme))
611 .map_err(|_| HttpClientError(format!("invalid scheme '{}'", parsed.scheme)))?;
612 let authority = format!("{}:{}", parsed.host, parsed.port);
613 request
614 .set_authority(Some(&authority))
615 .map_err(|_| HttpClientError(format!("invalid authority '{authority}'")))?;
616 request
617 .set_path_with_query(Some(&parsed.path_and_query))
618 .map_err(|_| HttpClientError(format!("invalid path '{}'", parsed.path_and_query)))?;
619
620 let outgoing_body = request
621 .body()
622 .map_err(|_| HttpClientError("outgoing-request body handle taken twice".to_string()))?;
623 if let Some(bytes) = body {
624 let mut out = outgoing_body
625 .write()
626 .map_err(|_| HttpClientError("outgoing-body write stream taken twice".to_string()))?;
627 out.write_all(bytes)
628 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
629 }
630 OutgoingBody::finish(outgoing_body, None)
631 .map_err(|e| HttpClientError(format!("failed to finish request body: {e:?}")))?;
632
633 let options = RequestOptions::new();
637 let timeout_ns = timeout_ms.saturating_mul(1_000_000);
638 let _ = options.set_connect_timeout(Some(timeout_ns));
639 let _ = options.set_first_byte_timeout(Some(timeout_ns));
640 let _ = options.set_between_bytes_timeout(Some(timeout_ns));
641
642 let future_response = outgoing_handler::handle(request, Some(options))
643 .map_err(|e| HttpClientError(format!("failed to dispatch request: {e:?}")))?;
644
645 future_response.subscribe().block();
646
647 let incoming_response = match future_response.get() {
648 Some(Ok(Ok(response))) => response,
649 Some(Ok(Err(code))) => return Err(HttpClientError(format!("{code:?}"))),
650 Some(Err(())) => {
651 return Err(HttpClientError("response already retrieved".to_string()))
652 }
653 None => {
654 return Err(HttpClientError(
655 "response not ready after blocking on it".to_string(),
656 ))
657 }
658 };
659
660 let status = incoming_response.status();
661 let resp_headers = incoming_response
662 .headers()
663 .entries()
664 .into_iter()
665 .map(|(name, value)| (name, String::from_utf8_lossy(&value).to_string()))
666 .collect();
667
668 let mut resp_body = Vec::new();
669 if !method.eq_ignore_ascii_case("HEAD") {
670 if let Ok(incoming_body) = incoming_response.consume() {
671 if let Ok(mut stream) = incoming_body.stream() {
672 let _ = stream.read_to_end(&mut resp_body);
673 }
674 }
675 }
676
677 Ok(Response { status, headers: resp_headers, body: resp_body })
678}
679
680#[cfg(target_arch = "wasm32")]
681fn to_wasi_method(method: &str) -> wasip2::http::types::Method {
682 use wasip2::http::types::Method;
683 match method.to_ascii_uppercase().as_str() {
684 "GET" => Method::Get,
685 "HEAD" => Method::Head,
686 "POST" => Method::Post,
687 "PUT" => Method::Put,
688 "DELETE" => Method::Delete,
689 "CONNECT" => Method::Connect,
690 "OPTIONS" => Method::Options,
691 "TRACE" => Method::Trace,
692 "PATCH" => Method::Patch,
693 other => Method::Other(other.to_string()),
694 }
695}
696
697pub struct Client {
705 timeout_ms: u64,
706 max_redirects: u8,
707 #[cfg(all(any(feature = "http-client", feature = "http2"), not(target_arch = "wasm32")))]
711 tls: Option<Arc<rustls::RootCertStore>>,
712}
713
714impl Client {
715 pub fn new() -> Self {
719 Self {
720 timeout_ms: 30_000,
721 max_redirects: 10,
722 #[cfg(all(any(feature = "http-client", feature = "http2"), not(target_arch = "wasm32")))]
723 tls: None,
724 }
725 }
726
727 #[cfg(all(any(feature = "http-client", feature = "http2"), not(target_arch = "wasm32")))]
732 pub fn ca_file(mut self, path: &str) -> Result<Self, HttpClientError> {
733 use rustls::pki_types::{pem::PemObject, CertificateDer};
734 let bytes = std::fs::read(path).map_err(|e| HttpClientError(format!("cannot read CA file '{path}': {e}")))?;
735 let mut roots = rustls::RootCertStore::empty();
736 for cert in CertificateDer::pem_slice_iter(&bytes) {
737 let cert = cert.map_err(|e| HttpClientError(format!("cannot parse CA file '{path}': {e}")))?;
738 roots.add(cert).map_err(|e| HttpClientError(format!("invalid CA certificate in '{path}': {e}")))?;
739 }
740 if roots.is_empty() {
741 return Err(HttpClientError(format!("no certificates in CA file '{path}'")));
742 }
743 self.tls = Some(Arc::new(roots));
744 Ok(self)
745 }
746
747 pub fn timeout_ms(mut self, ms: u64) -> Self {
749 self.timeout_ms = ms;
750 self
751 }
752
753 pub fn max_redirects(mut self, n: u8) -> Self {
755 self.max_redirects = n;
756 self
757 }
758
759 pub fn get(&self, url: &str) -> RequestBuilder<'_> {
761 self.request("GET", url)
762 }
763
764 pub fn post(&self, url: &str) -> RequestBuilder<'_> {
766 self.request("POST", url)
767 }
768
769 pub fn put(&self, url: &str) -> RequestBuilder<'_> {
771 self.request("PUT", url)
772 }
773
774 pub fn patch(&self, url: &str) -> RequestBuilder<'_> {
776 self.request("PATCH", url)
777 }
778
779 pub fn delete(&self, url: &str) -> RequestBuilder<'_> {
781 self.request("DELETE", url)
782 }
783
784 pub fn head(&self, url: &str) -> RequestBuilder<'_> {
786 self.request("HEAD", url)
787 }
788
789 pub fn request(&self, method: &str, url: &str) -> RequestBuilder<'_> {
791 RequestBuilder {
792 client: self,
793 method: method.to_uppercase(),
794 url: url.to_string(),
795 headers: Vec::new(),
796 body: None,
797 timeout_ms: None,
798 }
799 }
800}
801
802impl Default for Client {
803 fn default() -> Self {
804 Self::new()
805 }
806}
807
808pub struct RequestBuilder<'a> {
812 client: &'a Client,
813 method: String,
814 url: String,
815 headers: Vec<(String, String)>,
816 body: Option<Vec<u8>>,
817 timeout_ms: Option<u64>,
818}
819
820impl<'a> RequestBuilder<'a> {
821 pub fn header(mut self, name: &str, value: &str) -> Self {
823 self.headers.push((name.to_string(), value.to_string()));
824 self
825 }
826
827 pub fn body(mut self, bytes: Vec<u8>) -> Self {
829 self.body = Some(bytes);
830 self
831 }
832
833 pub fn body_text(mut self, s: &str) -> Self {
835 self.headers
836 .push(("Content-Type".to_string(), "text/plain".to_string()));
837 self.body = Some(s.as_bytes().to_vec());
838 self
839 }
840
841 pub fn body_json(mut self, s: &str) -> Self {
843 self.headers.push((
844 "Content-Type".to_string(),
845 "application/json".to_string(),
846 ));
847 self.body = Some(s.as_bytes().to_vec());
848 self
849 }
850
851 pub fn form(mut self, pairs: &[(&str, &str)]) -> Self {
867 self.headers.push((
868 "Content-Type".to_string(),
869 "application/x-www-form-urlencoded".to_string(),
870 ));
871 self.body = Some(form_urlencode(pairs).into_bytes());
872 self
873 }
874
875 pub fn timeout_ms(mut self, ms: u64) -> Self {
877 self.timeout_ms = Some(ms);
878 self
879 }
880
881 pub fn send(self) -> Result<Response, HttpClientError> {
886 #[cfg(not(target_arch = "wasm32"))]
889 let span = crate::otel::outbound_span(&self.method);
890 #[cfg(not(target_arch = "wasm32"))]
891 if let Some(span) = &span {
892 for (key, value) in crate::otel::http_client_attributes(&self.method, &self.url) {
893 span.set_attribute(key, value);
894 }
895 }
896 let mut headers = self.headers;
897 #[cfg(not(target_arch = "wasm32"))]
898 crate::otel::inject(&mut headers);
899 let result = Self::send_following(self.client, self.method, self.url, headers, self.body, self.timeout_ms);
900 #[cfg(not(target_arch = "wasm32"))]
901 if let Some(span) = &span {
902 span.finish_http_client(result.as_ref().map(|r| r.status()).map_err(|e| e.0.as_str()));
903 }
904 result
905 }
906
907 fn send_following(
908 client: &Client,
909 method: String,
910 url: String,
911 headers: Vec<(String, String)>,
912 body: Option<Vec<u8>>,
913 timeout_ms: Option<u64>,
914 ) -> Result<Response, HttpClientError> {
915 let timeout = timeout_ms.unwrap_or(client.timeout_ms);
916 let max_redirects = client.max_redirects;
917
918 let mut method = method;
919 let mut url = url;
920 let mut body = body;
921 let mut redirects = 0u8;
922
923 loop {
924 let parsed = ParsedUrl::parse(&url)?;
925 #[cfg(all(any(feature = "http-client", feature = "http2"), not(target_arch = "wasm32")))]
926 let resp = send_once(&method, &parsed, &headers, &body, timeout, client.tls.as_ref())?;
927 #[cfg(all(not(any(feature = "http-client", feature = "http2")), not(target_arch = "wasm32")))]
928 let resp = send_once(&method, &parsed, &headers, &body, timeout)?;
929 #[cfg(target_arch = "wasm32")]
930 let resp = send_once(&method, &parsed, &headers, &body, timeout, None)?;
931
932 if resp.is_redirect() && redirects < max_redirects {
933 let location = resp
934 .header("location")
935 .ok_or_else(|| HttpClientError("redirect with no Location header".into()))?
936 .to_string();
937 url = resolve_url(&url, &location);
938 redirects += 1;
939 if matches!(resp.status(), 301 | 302 | 303) {
940 method = "GET".to_string();
941 body = None;
942 }
943 continue;
944 }
945
946 return Ok(resp);
947 }
948 }
949}
950
951#[cfg(feature = "http2")]
954pub use async_impl::{AsyncClient, AsyncRequestBuilder};
955
956#[cfg(feature = "http2")]
957mod async_impl {
958 use super::{
959 build_request_bytes, decode_chunked, parse_status, resolve_url, HttpClientError,
960 ParsedUrl, Response,
961 };
962 use std::sync::Arc;
963 use tokio::io::{AsyncReadExt, AsyncWriteExt};
964
965 async fn async_tls_connect(
966 host: &str,
967 stream: tokio::net::TcpStream,
968 ) -> Result<tokio_rustls::client::TlsStream<tokio::net::TcpStream>, HttpClientError> {
969 use rustls::pki_types::ServerName;
970 use tokio_rustls::TlsConnector;
971
972 let root_store = rustls::RootCertStore::from_iter(
973 webpki_roots::TLS_SERVER_ROOTS.iter().cloned(),
974 );
975 let config = Arc::new(
976 crate::tls_client_builder()
977 .with_root_certificates(root_store)
978 .with_no_client_auth(),
979 );
980 let connector = TlsConnector::from(config);
981 let server_name = ServerName::try_from(host.to_string())
982 .map_err(|e| HttpClientError(format!("invalid hostname '{host}': {e}")))?;
983 connector
984 .connect(server_name, stream)
985 .await
986 .map_err(|e| HttpClientError(format!("TLS handshake failed: {e}")))
987 }
988
989 async fn async_read_response(
990 stream: &mut (impl AsyncReadExt + Unpin),
991 is_head: bool,
992 ) -> Result<Response, HttpClientError> {
993 let mut buf: Vec<u8> = Vec::with_capacity(8192);
994 let mut tmp = vec![0u8; 4096];
995
996 let header_end = loop {
997 let n = stream
998 .read(&mut tmp)
999 .await
1000 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
1001 if n == 0 {
1002 if buf.is_empty() {
1003 return Err(HttpClientError(
1004 "server closed connection without a response".into(),
1005 ));
1006 }
1007 break buf.len();
1008 }
1009 buf.extend_from_slice(&tmp[..n]);
1010 if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
1011 break pos + 4;
1012 }
1013 };
1014
1015 let header_block = std::str::from_utf8(&buf[..header_end])
1016 .map_err(|_| HttpClientError("response headers not UTF-8".into()))?;
1017
1018 let mut lines = header_block.lines();
1019 let status_line = lines
1020 .next()
1021 .ok_or_else(|| HttpClientError("empty response".into()))?;
1022 let status = parse_status(status_line)?;
1023
1024 let response_headers: Vec<(String, String)> = lines
1025 .filter_map(|line| {
1026 let mut parts = line.splitn(2, ':');
1027 let name = parts.next()?.trim().to_string();
1028 let value = parts.next()?.trim().to_string();
1029 if name.is_empty() { None } else { Some((name, value)) }
1030 })
1031 .collect();
1032
1033 let mut body = buf[header_end..].to_vec();
1034
1035 if !is_head {
1036 let transfer_encoding = response_headers
1037 .iter()
1038 .find(|(k, _)| k.to_lowercase() == "transfer-encoding")
1039 .map(|(_, v)| v.to_lowercase());
1040
1041 let content_length: Option<usize> = response_headers
1042 .iter()
1043 .find(|(k, _)| k.to_lowercase() == "content-length")
1044 .and_then(|(_, v)| v.trim().parse().ok());
1045
1046 if transfer_encoding
1047 .as_deref()
1048 .map(|te| te.contains("chunked"))
1049 .unwrap_or(false)
1050 {
1051 loop {
1052 let n = stream.read(&mut tmp).await
1053 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
1054 if n == 0 { break; }
1055 body.extend_from_slice(&tmp[..n]);
1056 }
1057 body = decode_chunked(&body)?;
1058 } else if let Some(len) = content_length {
1059 while body.len() < len {
1060 let n = stream.read(&mut tmp).await
1061 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
1062 if n == 0 { break; }
1063 body.extend_from_slice(&tmp[..n]);
1064 }
1065 body.truncate(len);
1066 } else {
1067 loop {
1068 let n = stream.read(&mut tmp).await
1069 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
1070 if n == 0 { break; }
1071 body.extend_from_slice(&tmp[..n]);
1072 }
1073 }
1074 } else {
1075 body.clear();
1076 }
1077
1078 Ok(Response { status, headers: response_headers, body })
1079 }
1080
1081 async fn async_send_once(
1082 method: &str,
1083 parsed: &ParsedUrl,
1084 headers: &[(String, String)],
1085 body: &Option<Vec<u8>>,
1086 timeout_ms: u64,
1087 ) -> Result<Response, HttpClientError> {
1088 use std::time::Duration;
1089 use tokio::net::TcpStream;
1090 use tokio::time::timeout;
1091
1092 let addr = format!("{}:{}", parsed.host, parsed.port);
1093 let dur = Duration::from_millis(timeout_ms);
1094 let request_bytes =
1095 build_request_bytes(method, &parsed.path_and_query, &parsed.host_header(), headers, body);
1096 let is_head = method.eq_ignore_ascii_case("HEAD");
1097
1098 let tcp = timeout(dur, TcpStream::connect(&addr))
1099 .await
1100 .map_err(|_| HttpClientError(format!("connect to '{addr}' timed out")))?
1101 .map_err(|e| HttpClientError(format!("connect to '{addr}' failed: {e}")))?;
1102
1103 if parsed.scheme == "https" {
1104 let tls_stream = timeout(dur, async_tls_connect(&parsed.host, tcp))
1105 .await
1106 .map_err(|_| HttpClientError("TLS handshake timed out".into()))??;
1107 let mut stream = tls_stream;
1108 timeout(dur, stream.write_all(&request_bytes))
1109 .await
1110 .map_err(|_| HttpClientError("write timed out".into()))?
1111 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
1112 return timeout(dur, async_read_response(&mut stream, is_head))
1113 .await
1114 .map_err(|_| HttpClientError("read timed out".into()))?;
1115 }
1116
1117 let mut stream = tcp;
1118 timeout(dur, stream.write_all(&request_bytes))
1119 .await
1120 .map_err(|_| HttpClientError("write timed out".into()))?
1121 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
1122 timeout(dur, async_read_response(&mut stream, is_head))
1123 .await
1124 .map_err(|_| HttpClientError("read timed out".into()))?
1125 }
1126
1127 pub struct AsyncClient {
1129 timeout_ms: u64,
1130 max_redirects: u8,
1131 }
1132
1133 impl AsyncClient {
1134 pub fn new() -> Self {
1136 Self {
1137 timeout_ms: 30_000,
1138 max_redirects: 10,
1139 }
1140 }
1141
1142 pub fn timeout_ms(mut self, ms: u64) -> Self {
1144 self.timeout_ms = ms;
1145 self
1146 }
1147
1148 pub fn max_redirects(mut self, n: u8) -> Self {
1150 self.max_redirects = n;
1151 self
1152 }
1153
1154 pub fn get(&self, url: &str) -> AsyncRequestBuilder<'_> {
1156 self.request("GET", url)
1157 }
1158
1159 pub fn post(&self, url: &str) -> AsyncRequestBuilder<'_> {
1161 self.request("POST", url)
1162 }
1163
1164 pub fn put(&self, url: &str) -> AsyncRequestBuilder<'_> {
1166 self.request("PUT", url)
1167 }
1168
1169 pub fn patch(&self, url: &str) -> AsyncRequestBuilder<'_> {
1171 self.request("PATCH", url)
1172 }
1173
1174 pub fn delete(&self, url: &str) -> AsyncRequestBuilder<'_> {
1176 self.request("DELETE", url)
1177 }
1178
1179 pub fn request(&self, method: &str, url: &str) -> AsyncRequestBuilder<'_> {
1181 AsyncRequestBuilder {
1182 client: self,
1183 method: method.to_uppercase(),
1184 url: url.to_string(),
1185 headers: Vec::new(),
1186 body: None,
1187 timeout_ms: None,
1188 }
1189 }
1190 }
1191
1192 impl Default for AsyncClient {
1193 fn default() -> Self {
1194 Self::new()
1195 }
1196 }
1197
1198 pub struct AsyncRequestBuilder<'a> {
1200 client: &'a AsyncClient,
1201 method: String,
1202 url: String,
1203 headers: Vec<(String, String)>,
1204 body: Option<Vec<u8>>,
1205 timeout_ms: Option<u64>,
1206 }
1207
1208 impl<'a> AsyncRequestBuilder<'a> {
1209 pub fn header(mut self, name: &str, value: &str) -> Self {
1211 self.headers.push((name.to_string(), value.to_string()));
1212 self
1213 }
1214
1215 pub fn body(mut self, bytes: Vec<u8>) -> Self {
1217 self.body = Some(bytes);
1218 self
1219 }
1220
1221 pub fn body_text(mut self, s: &str) -> Self {
1223 self.headers
1224 .push(("Content-Type".to_string(), "text/plain".to_string()));
1225 self.body = Some(s.as_bytes().to_vec());
1226 self
1227 }
1228
1229 pub fn body_json(mut self, s: &str) -> Self {
1231 self.headers.push((
1232 "Content-Type".to_string(),
1233 "application/json".to_string(),
1234 ));
1235 self.body = Some(s.as_bytes().to_vec());
1236 self
1237 }
1238
1239 pub fn form(mut self, pairs: &[(&str, &str)]) -> Self {
1242 self.headers.push((
1243 "Content-Type".to_string(),
1244 "application/x-www-form-urlencoded".to_string(),
1245 ));
1246 self.body = Some(super::form_urlencode(pairs).into_bytes());
1247 self
1248 }
1249
1250 pub fn timeout_ms(mut self, ms: u64) -> Self {
1252 self.timeout_ms = Some(ms);
1253 self
1254 }
1255
1256 pub async fn send(self) -> Result<Response, HttpClientError> {
1258 let mut span = crate::otel::detached_span(&self.method, crate::otel::SpanKind::Client);
1260 let mut headers = self.headers;
1261 if let Some(span) = &mut span {
1262 for (key, value) in crate::otel::http_client_attributes(&self.method, &self.url) {
1263 span.set_attribute(key, value);
1264 }
1265 span.enter(|| crate::otel::inject(&mut headers));
1266 }
1267 let result = Self::send_following(self.client, self.method, self.url, headers, self.body, self.timeout_ms).await;
1268 if let Some(span) = &mut span {
1269 span.finish_http_client(result.as_ref().map(|r| r.status()).map_err(|e| e.0.as_str()));
1270 }
1271 result
1272 }
1273
1274 async fn send_following(
1275 client: &AsyncClient,
1276 method: String,
1277 url: String,
1278 headers: Vec<(String, String)>,
1279 body: Option<Vec<u8>>,
1280 timeout_ms: Option<u64>,
1281 ) -> Result<Response, HttpClientError> {
1282 let timeout = timeout_ms.unwrap_or(client.timeout_ms);
1283 let max_redirects = client.max_redirects;
1284
1285 let mut method = method;
1286 let mut url = url;
1287 let mut body = body;
1288 let mut redirects = 0u8;
1289
1290 loop {
1291 let parsed = ParsedUrl::parse(&url)?;
1292 let resp = async_send_once(&method, &parsed, &headers, &body, timeout).await?;
1293
1294 if resp.is_redirect() && redirects < max_redirects {
1295 let location = resp
1296 .header("location")
1297 .ok_or_else(|| {
1298 HttpClientError("redirect with no Location header".into())
1299 })?
1300 .to_string();
1301 url = resolve_url(&url, &location);
1302 redirects += 1;
1303 if matches!(resp.status(), 301 | 302 | 303) {
1304 method = "GET".to_string();
1305 body = None;
1306 }
1307 continue;
1308 }
1309
1310 return Ok(resp);
1311 }
1312 }
1313 }
1314}