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