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) -> Result<rustls::StreamOwned<rustls::ClientConnection, TcpStream>, HttpClientError> {
495 use rustls::pki_types::ServerName;
496
497 let root_store =
498 rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
499 let config = Arc::new(
500 crate::tls_client_builder()
501 .with_root_certificates(root_store)
502 .with_no_client_auth(),
503 );
504 let server_name = ServerName::try_from(host.to_string())
505 .map_err(|e| HttpClientError(format!("invalid hostname '{host}': {e}")))?;
506 let conn = rustls::ClientConnection::new(config, server_name)
507 .map_err(|e| HttpClientError(e.to_string()))?;
508 Ok(rustls::StreamOwned::new(conn, tcp))
509}
510
511#[cfg(not(target_arch = "wasm32"))]
514fn send_once(
515 method: &str,
516 parsed: &ParsedUrl,
517 headers: &[(String, String)],
518 body: &Option<Vec<u8>>,
519 timeout_ms: u64,
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)?;
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) -> Result<Response, HttpClientError> {
595 use wasip2::http::outgoing_handler;
596 use wasip2::http::types::{Fields, OutgoingBody, OutgoingRequest, RequestOptions, Scheme};
597
598 let wasi_headers = Fields::new();
599 for (name, value) in headers {
600 let _ = wasi_headers.append(name, value.as_bytes());
601 }
602
603 let request = OutgoingRequest::new(wasi_headers);
604 request
605 .set_method(&to_wasi_method(method))
606 .map_err(|_| HttpClientError(format!("invalid method '{method}'")))?;
607 let scheme = if parsed.scheme == "https" { Scheme::Https } else { Scheme::Http };
608 request
609 .set_scheme(Some(&scheme))
610 .map_err(|_| HttpClientError(format!("invalid scheme '{}'", parsed.scheme)))?;
611 let authority = format!("{}:{}", parsed.host, parsed.port);
612 request
613 .set_authority(Some(&authority))
614 .map_err(|_| HttpClientError(format!("invalid authority '{authority}'")))?;
615 request
616 .set_path_with_query(Some(&parsed.path_and_query))
617 .map_err(|_| HttpClientError(format!("invalid path '{}'", parsed.path_and_query)))?;
618
619 let outgoing_body = request
620 .body()
621 .map_err(|_| HttpClientError("outgoing-request body handle taken twice".to_string()))?;
622 if let Some(bytes) = body {
623 let mut out = outgoing_body
624 .write()
625 .map_err(|_| HttpClientError("outgoing-body write stream taken twice".to_string()))?;
626 out.write_all(bytes)
627 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
628 }
629 OutgoingBody::finish(outgoing_body, None)
630 .map_err(|e| HttpClientError(format!("failed to finish request body: {e:?}")))?;
631
632 let options = RequestOptions::new();
636 let timeout_ns = timeout_ms.saturating_mul(1_000_000);
637 let _ = options.set_connect_timeout(Some(timeout_ns));
638 let _ = options.set_first_byte_timeout(Some(timeout_ns));
639 let _ = options.set_between_bytes_timeout(Some(timeout_ns));
640
641 let future_response = outgoing_handler::handle(request, Some(options))
642 .map_err(|e| HttpClientError(format!("failed to dispatch request: {e:?}")))?;
643
644 future_response.subscribe().block();
645
646 let incoming_response = match future_response.get() {
647 Some(Ok(Ok(response))) => response,
648 Some(Ok(Err(code))) => return Err(HttpClientError(format!("{code:?}"))),
649 Some(Err(())) => {
650 return Err(HttpClientError("response already retrieved".to_string()))
651 }
652 None => {
653 return Err(HttpClientError(
654 "response not ready after blocking on it".to_string(),
655 ))
656 }
657 };
658
659 let status = incoming_response.status();
660 let resp_headers = incoming_response
661 .headers()
662 .entries()
663 .into_iter()
664 .map(|(name, value)| (name, String::from_utf8_lossy(&value).to_string()))
665 .collect();
666
667 let mut resp_body = Vec::new();
668 if !method.eq_ignore_ascii_case("HEAD") {
669 if let Ok(incoming_body) = incoming_response.consume() {
670 if let Ok(mut stream) = incoming_body.stream() {
671 let _ = stream.read_to_end(&mut resp_body);
672 }
673 }
674 }
675
676 Ok(Response { status, headers: resp_headers, body: resp_body })
677}
678
679#[cfg(target_arch = "wasm32")]
680fn to_wasi_method(method: &str) -> wasip2::http::types::Method {
681 use wasip2::http::types::Method;
682 match method.to_ascii_uppercase().as_str() {
683 "GET" => Method::Get,
684 "HEAD" => Method::Head,
685 "POST" => Method::Post,
686 "PUT" => Method::Put,
687 "DELETE" => Method::Delete,
688 "CONNECT" => Method::Connect,
689 "OPTIONS" => Method::Options,
690 "TRACE" => Method::Trace,
691 "PATCH" => Method::Patch,
692 other => Method::Other(other.to_string()),
693 }
694}
695
696pub struct Client {
704 timeout_ms: u64,
705 max_redirects: u8,
706}
707
708impl Client {
709 pub fn new() -> Self {
713 Self {
714 timeout_ms: 30_000,
715 max_redirects: 10,
716 }
717 }
718
719 pub fn timeout_ms(mut self, ms: u64) -> Self {
721 self.timeout_ms = ms;
722 self
723 }
724
725 pub fn max_redirects(mut self, n: u8) -> Self {
727 self.max_redirects = n;
728 self
729 }
730
731 pub fn get(&self, url: &str) -> RequestBuilder<'_> {
733 self.request("GET", url)
734 }
735
736 pub fn post(&self, url: &str) -> RequestBuilder<'_> {
738 self.request("POST", url)
739 }
740
741 pub fn put(&self, url: &str) -> RequestBuilder<'_> {
743 self.request("PUT", url)
744 }
745
746 pub fn patch(&self, url: &str) -> RequestBuilder<'_> {
748 self.request("PATCH", url)
749 }
750
751 pub fn delete(&self, url: &str) -> RequestBuilder<'_> {
753 self.request("DELETE", url)
754 }
755
756 pub fn head(&self, url: &str) -> RequestBuilder<'_> {
758 self.request("HEAD", url)
759 }
760
761 pub fn request(&self, method: &str, url: &str) -> RequestBuilder<'_> {
763 RequestBuilder {
764 client: self,
765 method: method.to_uppercase(),
766 url: url.to_string(),
767 headers: Vec::new(),
768 body: None,
769 timeout_ms: None,
770 }
771 }
772}
773
774impl Default for Client {
775 fn default() -> Self {
776 Self::new()
777 }
778}
779
780pub struct RequestBuilder<'a> {
784 client: &'a Client,
785 method: String,
786 url: String,
787 headers: Vec<(String, String)>,
788 body: Option<Vec<u8>>,
789 timeout_ms: Option<u64>,
790}
791
792impl<'a> RequestBuilder<'a> {
793 pub fn header(mut self, name: &str, value: &str) -> Self {
795 self.headers.push((name.to_string(), value.to_string()));
796 self
797 }
798
799 pub fn body(mut self, bytes: Vec<u8>) -> Self {
801 self.body = Some(bytes);
802 self
803 }
804
805 pub fn body_text(mut self, s: &str) -> Self {
807 self.headers
808 .push(("Content-Type".to_string(), "text/plain".to_string()));
809 self.body = Some(s.as_bytes().to_vec());
810 self
811 }
812
813 pub fn body_json(mut self, s: &str) -> Self {
815 self.headers.push((
816 "Content-Type".to_string(),
817 "application/json".to_string(),
818 ));
819 self.body = Some(s.as_bytes().to_vec());
820 self
821 }
822
823 pub fn form(mut self, pairs: &[(&str, &str)]) -> Self {
839 self.headers.push((
840 "Content-Type".to_string(),
841 "application/x-www-form-urlencoded".to_string(),
842 ));
843 self.body = Some(form_urlencode(pairs).into_bytes());
844 self
845 }
846
847 pub fn timeout_ms(mut self, ms: u64) -> Self {
849 self.timeout_ms = Some(ms);
850 self
851 }
852
853 pub fn send(self) -> Result<Response, HttpClientError> {
858 #[cfg(not(target_arch = "wasm32"))]
861 let span = crate::otel::outbound_span(&self.method);
862 #[cfg(not(target_arch = "wasm32"))]
863 if let Some(span) = &span {
864 for (key, value) in crate::otel::http_client_attributes(&self.method, &self.url) {
865 span.set_attribute(key, value);
866 }
867 }
868 let mut headers = self.headers;
869 #[cfg(not(target_arch = "wasm32"))]
870 crate::otel::inject(&mut headers);
871 let result = Self::send_following(self.client, self.method, self.url, headers, self.body, self.timeout_ms);
872 #[cfg(not(target_arch = "wasm32"))]
873 if let Some(span) = &span {
874 span.finish_http_client(result.as_ref().map(|r| r.status()).map_err(|e| e.0.as_str()));
875 }
876 result
877 }
878
879 fn send_following(
880 client: &Client,
881 method: String,
882 url: String,
883 headers: Vec<(String, String)>,
884 body: Option<Vec<u8>>,
885 timeout_ms: Option<u64>,
886 ) -> Result<Response, HttpClientError> {
887 let timeout = timeout_ms.unwrap_or(client.timeout_ms);
888 let max_redirects = client.max_redirects;
889
890 let mut method = method;
891 let mut url = url;
892 let mut body = body;
893 let mut redirects = 0u8;
894
895 loop {
896 let parsed = ParsedUrl::parse(&url)?;
897 let resp = send_once(&method, &parsed, &headers, &body, timeout)?;
898
899 if resp.is_redirect() && redirects < max_redirects {
900 let location = resp
901 .header("location")
902 .ok_or_else(|| HttpClientError("redirect with no Location header".into()))?
903 .to_string();
904 url = resolve_url(&url, &location);
905 redirects += 1;
906 if matches!(resp.status(), 301 | 302 | 303) {
907 method = "GET".to_string();
908 body = None;
909 }
910 continue;
911 }
912
913 return Ok(resp);
914 }
915 }
916}
917
918#[cfg(feature = "http2")]
921pub use async_impl::{AsyncClient, AsyncRequestBuilder};
922
923#[cfg(feature = "http2")]
924mod async_impl {
925 use super::{
926 build_request_bytes, decode_chunked, parse_status, resolve_url, HttpClientError,
927 ParsedUrl, Response,
928 };
929 use std::sync::Arc;
930 use tokio::io::{AsyncReadExt, AsyncWriteExt};
931
932 async fn async_tls_connect(
933 host: &str,
934 stream: tokio::net::TcpStream,
935 ) -> Result<tokio_rustls::client::TlsStream<tokio::net::TcpStream>, HttpClientError> {
936 use rustls::pki_types::ServerName;
937 use tokio_rustls::TlsConnector;
938
939 let root_store = rustls::RootCertStore::from_iter(
940 webpki_roots::TLS_SERVER_ROOTS.iter().cloned(),
941 );
942 let config = Arc::new(
943 crate::tls_client_builder()
944 .with_root_certificates(root_store)
945 .with_no_client_auth(),
946 );
947 let connector = TlsConnector::from(config);
948 let server_name = ServerName::try_from(host.to_string())
949 .map_err(|e| HttpClientError(format!("invalid hostname '{host}': {e}")))?;
950 connector
951 .connect(server_name, stream)
952 .await
953 .map_err(|e| HttpClientError(format!("TLS handshake failed: {e}")))
954 }
955
956 async fn async_read_response(
957 stream: &mut (impl AsyncReadExt + Unpin),
958 is_head: bool,
959 ) -> Result<Response, HttpClientError> {
960 let mut buf: Vec<u8> = Vec::with_capacity(8192);
961 let mut tmp = vec![0u8; 4096];
962
963 let header_end = loop {
964 let n = stream
965 .read(&mut tmp)
966 .await
967 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
968 if n == 0 {
969 if buf.is_empty() {
970 return Err(HttpClientError(
971 "server closed connection without a response".into(),
972 ));
973 }
974 break buf.len();
975 }
976 buf.extend_from_slice(&tmp[..n]);
977 if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
978 break pos + 4;
979 }
980 };
981
982 let header_block = std::str::from_utf8(&buf[..header_end])
983 .map_err(|_| HttpClientError("response headers not UTF-8".into()))?;
984
985 let mut lines = header_block.lines();
986 let status_line = lines
987 .next()
988 .ok_or_else(|| HttpClientError("empty response".into()))?;
989 let status = parse_status(status_line)?;
990
991 let response_headers: Vec<(String, String)> = lines
992 .filter_map(|line| {
993 let mut parts = line.splitn(2, ':');
994 let name = parts.next()?.trim().to_string();
995 let value = parts.next()?.trim().to_string();
996 if name.is_empty() { None } else { Some((name, value)) }
997 })
998 .collect();
999
1000 let mut body = buf[header_end..].to_vec();
1001
1002 if !is_head {
1003 let transfer_encoding = response_headers
1004 .iter()
1005 .find(|(k, _)| k.to_lowercase() == "transfer-encoding")
1006 .map(|(_, v)| v.to_lowercase());
1007
1008 let content_length: Option<usize> = response_headers
1009 .iter()
1010 .find(|(k, _)| k.to_lowercase() == "content-length")
1011 .and_then(|(_, v)| v.trim().parse().ok());
1012
1013 if transfer_encoding
1014 .as_deref()
1015 .map(|te| te.contains("chunked"))
1016 .unwrap_or(false)
1017 {
1018 loop {
1019 let n = stream.read(&mut tmp).await
1020 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
1021 if n == 0 { break; }
1022 body.extend_from_slice(&tmp[..n]);
1023 }
1024 body = decode_chunked(&body)?;
1025 } else if let Some(len) = content_length {
1026 while body.len() < len {
1027 let n = stream.read(&mut tmp).await
1028 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
1029 if n == 0 { break; }
1030 body.extend_from_slice(&tmp[..n]);
1031 }
1032 body.truncate(len);
1033 } else {
1034 loop {
1035 let n = stream.read(&mut tmp).await
1036 .map_err(|e| HttpClientError(format!("read error: {e}")))?;
1037 if n == 0 { break; }
1038 body.extend_from_slice(&tmp[..n]);
1039 }
1040 }
1041 } else {
1042 body.clear();
1043 }
1044
1045 Ok(Response { status, headers: response_headers, body })
1046 }
1047
1048 async fn async_send_once(
1049 method: &str,
1050 parsed: &ParsedUrl,
1051 headers: &[(String, String)],
1052 body: &Option<Vec<u8>>,
1053 timeout_ms: u64,
1054 ) -> Result<Response, HttpClientError> {
1055 use std::time::Duration;
1056 use tokio::net::TcpStream;
1057 use tokio::time::timeout;
1058
1059 let addr = format!("{}:{}", parsed.host, parsed.port);
1060 let dur = Duration::from_millis(timeout_ms);
1061 let request_bytes =
1062 build_request_bytes(method, &parsed.path_and_query, &parsed.host_header(), headers, body);
1063 let is_head = method.eq_ignore_ascii_case("HEAD");
1064
1065 let tcp = timeout(dur, TcpStream::connect(&addr))
1066 .await
1067 .map_err(|_| HttpClientError(format!("connect to '{addr}' timed out")))?
1068 .map_err(|e| HttpClientError(format!("connect to '{addr}' failed: {e}")))?;
1069
1070 if parsed.scheme == "https" {
1071 let tls_stream = timeout(dur, async_tls_connect(&parsed.host, tcp))
1072 .await
1073 .map_err(|_| HttpClientError("TLS handshake timed out".into()))??;
1074 let mut stream = tls_stream;
1075 timeout(dur, stream.write_all(&request_bytes))
1076 .await
1077 .map_err(|_| HttpClientError("write timed out".into()))?
1078 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
1079 return timeout(dur, async_read_response(&mut stream, is_head))
1080 .await
1081 .map_err(|_| HttpClientError("read timed out".into()))?;
1082 }
1083
1084 let mut stream = tcp;
1085 timeout(dur, stream.write_all(&request_bytes))
1086 .await
1087 .map_err(|_| HttpClientError("write timed out".into()))?
1088 .map_err(|e| HttpClientError(format!("write error: {e}")))?;
1089 timeout(dur, async_read_response(&mut stream, is_head))
1090 .await
1091 .map_err(|_| HttpClientError("read timed out".into()))?
1092 }
1093
1094 pub struct AsyncClient {
1096 timeout_ms: u64,
1097 max_redirects: u8,
1098 }
1099
1100 impl AsyncClient {
1101 pub fn new() -> Self {
1103 Self {
1104 timeout_ms: 30_000,
1105 max_redirects: 10,
1106 }
1107 }
1108
1109 pub fn timeout_ms(mut self, ms: u64) -> Self {
1111 self.timeout_ms = ms;
1112 self
1113 }
1114
1115 pub fn max_redirects(mut self, n: u8) -> Self {
1117 self.max_redirects = n;
1118 self
1119 }
1120
1121 pub fn get(&self, url: &str) -> AsyncRequestBuilder<'_> {
1123 self.request("GET", url)
1124 }
1125
1126 pub fn post(&self, url: &str) -> AsyncRequestBuilder<'_> {
1128 self.request("POST", url)
1129 }
1130
1131 pub fn put(&self, url: &str) -> AsyncRequestBuilder<'_> {
1133 self.request("PUT", url)
1134 }
1135
1136 pub fn patch(&self, url: &str) -> AsyncRequestBuilder<'_> {
1138 self.request("PATCH", url)
1139 }
1140
1141 pub fn delete(&self, url: &str) -> AsyncRequestBuilder<'_> {
1143 self.request("DELETE", url)
1144 }
1145
1146 pub fn request(&self, method: &str, url: &str) -> AsyncRequestBuilder<'_> {
1148 AsyncRequestBuilder {
1149 client: self,
1150 method: method.to_uppercase(),
1151 url: url.to_string(),
1152 headers: Vec::new(),
1153 body: None,
1154 timeout_ms: None,
1155 }
1156 }
1157 }
1158
1159 impl Default for AsyncClient {
1160 fn default() -> Self {
1161 Self::new()
1162 }
1163 }
1164
1165 pub struct AsyncRequestBuilder<'a> {
1167 client: &'a AsyncClient,
1168 method: String,
1169 url: String,
1170 headers: Vec<(String, String)>,
1171 body: Option<Vec<u8>>,
1172 timeout_ms: Option<u64>,
1173 }
1174
1175 impl<'a> AsyncRequestBuilder<'a> {
1176 pub fn header(mut self, name: &str, value: &str) -> Self {
1178 self.headers.push((name.to_string(), value.to_string()));
1179 self
1180 }
1181
1182 pub fn body(mut self, bytes: Vec<u8>) -> Self {
1184 self.body = Some(bytes);
1185 self
1186 }
1187
1188 pub fn body_text(mut self, s: &str) -> Self {
1190 self.headers
1191 .push(("Content-Type".to_string(), "text/plain".to_string()));
1192 self.body = Some(s.as_bytes().to_vec());
1193 self
1194 }
1195
1196 pub fn body_json(mut self, s: &str) -> Self {
1198 self.headers.push((
1199 "Content-Type".to_string(),
1200 "application/json".to_string(),
1201 ));
1202 self.body = Some(s.as_bytes().to_vec());
1203 self
1204 }
1205
1206 pub fn form(mut self, pairs: &[(&str, &str)]) -> Self {
1209 self.headers.push((
1210 "Content-Type".to_string(),
1211 "application/x-www-form-urlencoded".to_string(),
1212 ));
1213 self.body = Some(super::form_urlencode(pairs).into_bytes());
1214 self
1215 }
1216
1217 pub fn timeout_ms(mut self, ms: u64) -> Self {
1219 self.timeout_ms = Some(ms);
1220 self
1221 }
1222
1223 pub async fn send(self) -> Result<Response, HttpClientError> {
1225 let mut span = crate::otel::detached_span(&self.method, crate::otel::SpanKind::Client);
1227 let mut headers = self.headers;
1228 if let Some(span) = &mut span {
1229 for (key, value) in crate::otel::http_client_attributes(&self.method, &self.url) {
1230 span.set_attribute(key, value);
1231 }
1232 span.enter(|| crate::otel::inject(&mut headers));
1233 }
1234 let result = Self::send_following(self.client, self.method, self.url, headers, self.body, self.timeout_ms).await;
1235 if let Some(span) = &mut span {
1236 span.finish_http_client(result.as_ref().map(|r| r.status()).map_err(|e| e.0.as_str()));
1237 }
1238 result
1239 }
1240
1241 async fn send_following(
1242 client: &AsyncClient,
1243 method: String,
1244 url: String,
1245 headers: Vec<(String, String)>,
1246 body: Option<Vec<u8>>,
1247 timeout_ms: Option<u64>,
1248 ) -> Result<Response, HttpClientError> {
1249 let timeout = timeout_ms.unwrap_or(client.timeout_ms);
1250 let max_redirects = client.max_redirects;
1251
1252 let mut method = method;
1253 let mut url = url;
1254 let mut body = body;
1255 let mut redirects = 0u8;
1256
1257 loop {
1258 let parsed = ParsedUrl::parse(&url)?;
1259 let resp = async_send_once(&method, &parsed, &headers, &body, timeout).await?;
1260
1261 if resp.is_redirect() && redirects < max_redirects {
1262 let location = resp
1263 .header("location")
1264 .ok_or_else(|| {
1265 HttpClientError("redirect with no Location header".into())
1266 })?
1267 .to_string();
1268 url = resolve_url(&url, &location);
1269 redirects += 1;
1270 if matches!(resp.status(), 301 | 302 | 303) {
1271 method = "GET".to_string();
1272 body = None;
1273 }
1274 continue;
1275 }
1276
1277 return Ok(resp);
1278 }
1279 }
1280 }
1281}