go_http/
client.rs

1// SPDX-License-Identifier: Apache-2.0
2
3/// Client, Transport, and RoundTripper — port of Go's net/http client.
4use std::collections::{HashMap, VecDeque};
5use std::io::{self, Read, Write};
6use std::sync::{Arc, Mutex};
7use std::time::Duration;
8
9use go_lib::context::{with_timeout, CancelFn, Context};
10use go_lib::net::TcpStream;
11use url::Url;
12
13use crate::cookie::{Cookie, CookieJar};
14use crate::error::HttpError;
15use crate::header::Header;
16use crate::parse::response::{read_response, ParsedResponse};
17use crate::parse::transfer::Body;
18use crate::request::Request;
19use crate::response::Response;
20
21// ---------------------------------------------------------------------------
22// RoundTripper — port of Go's http.RoundTripper
23// ---------------------------------------------------------------------------
24
25/// The low-level interface for executing a single HTTP request.
26/// Port of Go's `http.RoundTripper`.
27pub trait RoundTripper: Send + Sync {
28    fn round_trip(&self, req: Request) -> Result<Response, HttpError>;
29}
30
31// ---------------------------------------------------------------------------
32// Transport — default RoundTripper with connection pooling
33// ---------------------------------------------------------------------------
34
35/// A connection pool entry: an idle `TcpStream` ready for reuse.
36struct IdleConn {
37    stream: TcpStream,
38}
39
40/// Default `RoundTripper` with per-host idle connection pooling.
41/// Port of Go's `http.Transport`.
42pub struct Transport {
43    pub max_idle_conns_per_host: usize,
44    pub idle_conn_timeout:       Option<Duration>,
45    pub dial_timeout:            Option<Duration>,
46    /// TLS client configuration for HTTPS requests.
47    /// `None` uses the default Mozilla root store (via `webpki-roots`).
48    pub tls_config: Option<Arc<rustls::ClientConfig>>,
49    /// Idle connection pool keyed by `"host:port"`.
50    pool: Mutex<HashMap<String, VecDeque<IdleConn>>>,
51}
52
53impl Transport {
54    pub fn new() -> Self {
55        Self {
56            max_idle_conns_per_host: 10,
57            idle_conn_timeout:       Some(Duration::from_secs(90)),
58            dial_timeout:            Some(Duration::from_secs(30)),
59            tls_config:              None,
60            pool:                    Mutex::new(HashMap::new()),
61        }
62    }
63
64    /// Acquire an idle connection for `host_port`, or dial a new one.
65    fn acquire(&self, host_port: &str) -> io::Result<TcpStream> {
66        // Try pool first.
67        if let Some(conn) = self
68            .pool
69            .lock()
70            .unwrap()
71            .get_mut(host_port)
72            .and_then(|q| q.pop_front())
73        {
74            return Ok(conn.stream);
75        }
76        // Dial a new connection.
77        TcpStream::connect(host_port)
78    }
79
80    /// Return a connection to the pool for reuse.
81    fn release(&self, host_port: &str, stream: TcpStream) {
82        let mut pool = self.pool.lock().unwrap();
83        let queue = pool.entry(host_port.to_owned()).or_default();
84        if queue.len() < self.max_idle_conns_per_host {
85            queue.push_back(IdleConn { stream });
86        }
87        // If over the limit we simply drop the stream (closes the fd).
88    }
89}
90
91impl Default for Transport {
92    fn default() -> Self {
93        Self::new()
94    }
95}
96
97impl RoundTripper for Transport {
98    fn round_trip(&self, mut req: Request) -> Result<Response, HttpError> {
99        let scheme    = req.url.scheme();
100        let is_https  = scheme == "https";
101        let host      = req.url.host_str().unwrap_or("localhost");
102        let port      = req.url.port_or_known_default()
103            .unwrap_or(if is_https { 443 } else { 80 });
104        let host_port = format!("{host}:{port}");
105
106        let stream = self.acquire(&host_port).map_err(HttpError::Io)?;
107
108        if is_https {
109            // ── HTTPS path ────────────────────────────────────────────────────
110            let tls_cfg = match &self.tls_config {
111                Some(c) => Arc::clone(c),
112                None    => crate::tls::default_client_config(),
113            };
114            let server_name = rustls::pki_types::ServerName::try_from(host.to_owned())
115                .map_err(|e| HttpError::Tls(e.to_string()))?;
116            let client_conn = rustls::ClientConnection::new(tls_cfg, server_name)
117                .map_err(|e| HttpError::Tls(e.to_string()))?;
118            let mut tls = rustls::StreamOwned::new(client_conn, stream);
119
120            send_request(&mut tls, &mut req)?;
121
122            // Lend `tls` to the response parser via a raw-pointer read wrapper.
123            // Safety: `tls` outlives the parser and the response body within this
124            // call frame; there is no concurrent access.
125            let read_ptr: *mut dyn Read = &mut tls as &mut dyn Read as *mut dyn Read;
126            let parsed = read_response(
127                RawRead(read_ptr),
128                Some(req.method.as_str()),
129                crate::parse::request::DEFAULT_MAX_HEADER_BYTES,
130            )?;
131            // TLS connections are not pooled (no try_clone equivalent).
132            Ok(parsed_response_to_response(parsed))
133        } else {
134            // ── HTTP path ─────────────────────────────────────────────────────
135            let mut stream = stream;
136            send_request(&mut stream, &mut req)?;
137
138            let mut parsed = read_response(
139                stream.try_clone().map_err(HttpError::Io)?,
140                Some(req.method.as_str()),
141                crate::parse::request::DEFAULT_MAX_HEADER_BYTES,
142            )?;
143
144            let keep_alive = is_keep_alive_parsed(&parsed, req.proto_minor);
145
146            // Buffer the body into memory before releasing the stream to the
147            // pool.  The body is backed by a try_clone() of `stream`; if we
148            // release `stream` first and the caller drops the Response without
149            // reading the body, Body::drain() would race against the next
150            // request's reads on the same underlying socket (clone and
151            // original share one kernel file description) → SIGSEGV on
152            // Linux/epoll.  Buffering here fully drains the socket before the
153            // connection is reused, and replaces the network-backed body with
154            // an in-memory Cursor the caller can read freely.
155            if keep_alive {
156                let bytes = parsed.body.read_to_vec().map_err(|_| HttpError::BodyRead)?;
157                parsed.body = Body::Unbounded(Box::new(io::Cursor::new(bytes)));
158                self.release(&host_port, stream);
159            }
160
161            Ok(parsed_response_to_response(parsed))
162        }
163    }
164}
165
166// ---------------------------------------------------------------------------
167// RawRead — lends a mutable reference as a Send + 'static Read
168// ---------------------------------------------------------------------------
169
170/// Raw-pointer wrapper that gives `read_response` a `Read + Send + 'static`
171/// view of a `TLS StreamOwned` (or any `impl Read`) without moving it.
172///
173/// # Safety
174/// The caller must ensure the pointed-to value lives at least as long as
175/// this `RawRead` is alive and is not concurrently accessed.
176struct RawRead(*mut dyn Read);
177unsafe impl Send for RawRead {}
178impl Read for RawRead {
179    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
180        unsafe { (*self.0).read(buf) }
181    }
182}
183
184// ---------------------------------------------------------------------------
185// Client
186// ---------------------------------------------------------------------------
187
188/// An HTTP client.  Mirrors Go's `http.Client`.
189pub struct Client {
190    /// Transport used for request execution.
191    pub transport:    Arc<dyn RoundTripper>,
192    /// Per-request timeout; `None` means no timeout.
193    pub timeout:      Option<Duration>,
194    /// Maximum number of redirects to follow (default 10, matching Go).
195    pub max_redirects: usize,
196    /// Optional cookie jar.
197    pub jar: Option<Arc<dyn CookieJar>>,
198}
199
200impl Client {
201    /// Create a client using the default `Transport`.
202    pub fn new() -> Self {
203        Self {
204            transport:     Arc::new(Transport::new()),
205            timeout:       None,
206            max_redirects: 10,
207            jar:           None,
208        }
209    }
210
211    // ── Convenience methods ───────────────────────────────────────────────
212
213    /// Issue a GET request.  Port of Go's `(*Client).Get`.
214    pub fn get(&self, url: &str) -> Result<Response, HttpError> {
215        let req = Request::new("GET", url, None)?;
216        self.do_request(req)
217    }
218
219    /// Issue a POST request with the given content type and body.
220    /// Port of Go's `(*Client).Post`.
221    pub fn post(
222        &self,
223        url:          &str,
224        content_type: &str,
225        body:         Body,
226    ) -> Result<Response, HttpError> {
227        let mut req = Request::new("POST", url, Some(body))?;
228        req.header.set("Content-Type", content_type);
229        self.do_request(req)
230    }
231
232    /// Issue a POST request with `application/x-www-form-urlencoded` body.
233    /// Port of Go's `(*Client).PostForm`.
234    pub fn post_form(
235        &self,
236        url:    &str,
237        values: &[(&str, &str)],
238    ) -> Result<Response, HttpError> {
239        let encoded = url_encode(values);
240        let body = Body::Unbounded(Box::new(io::Cursor::new(encoded.into_bytes())));
241        self.post(url, "application/x-www-form-urlencoded", body)
242    }
243
244    /// Issue a HEAD request.  Port of Go's `(*Client).Head`.
245    pub fn head(&self, url: &str) -> Result<Response, HttpError> {
246        let req = Request::new("HEAD", url, None)?;
247        self.do_request(req)
248    }
249
250    // ── Core request execution ────────────────────────────────────────────
251
252    /// Execute `req`, following redirects and attaching cookies.
253    /// Port of Go's `(*Client).Do`.
254    pub fn do_request(&self, req: Request) -> Result<Response, HttpError> {
255        // Wrap in a context deadline if a timeout is configured.
256        let (_cancel, ctx_req) = apply_timeout(&req, self.timeout);
257        let mut req = if let Some(ctx) = ctx_req { req.with_context(ctx) } else { req };
258
259        // Attach cookies from the jar for the initial URL.
260        if let Some(jar) = &self.jar {
261            attach_cookies(&mut req, jar.as_ref());
262        }
263
264        let mut redirects = 0usize;
265
266        loop {
267            let method = req.method.clone();
268            let url    = req.url.clone();
269
270            let mut resp = self.transport.round_trip(req)?;
271
272            // Store cookies from the response.
273            if let Some(jar) = &self.jar {
274                store_cookies(&url, &resp.header, jar.as_ref());
275            }
276
277            // ── Redirect handling ─────────────────────────────────────────
278            let status = resp.status;
279            if !is_redirect(status) {
280                return Ok(resp);
281            }
282
283            if redirects >= self.max_redirects {
284                return Err(HttpError::TooManyRedirects);
285            }
286            redirects += 1;
287
288            let location = resp
289                .header
290                .get("Location")
291                .ok_or_else(|| HttpError::InvalidUrl("redirect with no Location".into()))?
292                .to_owned();
293
294            // Drain the redirect body so the connection can be reused.
295            let _ = resp.body_bytes();
296
297            // Resolve the redirect URL against the original.
298            let new_url = resolve_url(&url, &location)?;
299
300            // POST → GET on 301/302/303 (matching Go semantics).
301            let new_method = match status {
302                301 | 302 | 303 => {
303                    if method == "POST" { "GET".to_owned() } else { method }
304                }
305                _ => method,
306            };
307
308            let body = if new_method == "GET" || new_method == "HEAD" {
309                None
310            } else {
311                None // body consumed; caller must handle 307/308 with body separately
312            };
313
314            let mut new_req = Request::new(&new_method, new_url.as_str(), body)?;
315            // Forward safe headers; strip Authorization on cross-origin redirects.
316            forward_headers(&mut new_req.header, &resp.header, same_origin(&url, &new_url));
317
318            if let Some(jar) = &self.jar {
319                attach_cookies(&mut new_req, jar.as_ref());
320            }
321
322            req = new_req;
323        }
324    }
325}
326
327impl Default for Client {
328    fn default() -> Self {
329        Self::new()
330    }
331}
332
333// ---------------------------------------------------------------------------
334// Package-level free functions — port of Go's http.Get / http.Post etc.
335// ---------------------------------------------------------------------------
336
337/// Global default client, mirroring Go's `http.DefaultClient`.
338fn default_client() -> &'static Client {
339    use std::sync::OnceLock;
340    static DEFAULT: OnceLock<Client> = OnceLock::new();
341    DEFAULT.get_or_init(Client::new)
342}
343
344/// Issue a GET using the default client.  Port of Go's `http.Get`.
345pub fn get(url: &str) -> Result<Response, HttpError> {
346    default_client().get(url)
347}
348
349/// Issue a POST using the default client.  Port of Go's `http.Post`.
350pub fn post(url: &str, content_type: &str, body: Body) -> Result<Response, HttpError> {
351    default_client().post(url, content_type, body)
352}
353
354/// Issue a POST form using the default client.  Port of Go's `http.PostForm`.
355pub fn post_form(url: &str, values: &[(&str, &str)]) -> Result<Response, HttpError> {
356    default_client().post_form(url, values)
357}
358
359/// Issue a HEAD using the default client.
360pub fn head(url: &str) -> Result<Response, HttpError> {
361    default_client().head(url)
362}
363
364// ---------------------------------------------------------------------------
365// Internal helpers
366// ---------------------------------------------------------------------------
367
368/// Serialize a `Request` (headers + body) to `w`.
369fn send_request(w: &mut impl Write, req: &mut Request) -> Result<(), HttpError> {
370    req.write_header_to(w)?;
371    // Write body bytes if present (POST/PUT/PATCH).
372    if let Some(body) = req.body.take() {
373        use std::io::Read;
374        let mut body = body;
375        let mut buf = [0u8; 8192];
376        loop {
377            let n = body.read(&mut buf).map_err(|_| HttpError::BodyRead)?;
378            if n == 0 { break; }
379            w.write_all(&buf[..n])?;
380        }
381    }
382    Ok(())
383}
384
385/// True if a `ParsedResponse` should be treated as keep-alive.
386fn is_keep_alive_parsed(resp: &ParsedResponse, req_minor: u8) -> bool {
387    let conn = resp.header.get("Connection").unwrap_or("").to_ascii_lowercase();
388    if conn.contains("close") { return false; }
389    if req_minor == 0 { conn.contains("keep-alive") } else { true }
390}
391
392/// Convert a `ParsedResponse` into the public `Response` type.
393fn parsed_response_to_response(p: ParsedResponse) -> Response {
394    Response {
395        status:            p.status,
396        status_text:       p.status_text,
397        proto:             p.proto,
398        proto_major:       p.proto_major,
399        proto_minor:       p.proto_minor,
400        header:            p.header,
401        body:              match p.body {
402            Body::Empty => None,
403            other       => Some(other),
404        },
405        content_length:    p.content_length,
406        transfer_encoding: p.transfer_encoding,
407        trailer:           Header::new(),
408    }
409}
410
411/// Copy safe headers from `src` into `dst`; strip Authorization on
412/// cross-origin redirects.
413fn forward_headers(dst: &mut Header, src: &Header, same_origin: bool) {
414    for (name, values) in src.iter() {
415        // Never forward hop-by-hop headers.
416        let lower = name.to_ascii_lowercase();
417        if matches!(
418            lower.as_str(),
419            "connection" | "keep-alive" | "proxy-authenticate"
420                | "proxy-authorization" | "te" | "trailers"
421                | "transfer-encoding" | "upgrade"
422        ) {
423            continue;
424        }
425        // Strip Authorization on cross-origin redirects (Go behaviour).
426        if !same_origin && lower == "authorization" {
427            continue;
428        }
429        for v in values {
430            dst.add(name, v.as_str());
431        }
432    }
433}
434
435/// True if `status` is a redirect code.
436fn is_redirect(status: u16) -> bool {
437    matches!(status, 301 | 302 | 303 | 307 | 308)
438}
439
440/// Resolve `location` (possibly relative) against `base`.
441fn resolve_url(base: &Url, location: &str) -> Result<Url, HttpError> {
442    if location.starts_with("http://") || location.starts_with("https://") {
443        Url::parse(location).map_err(|e| HttpError::InvalidUrl(e.to_string()))
444    } else {
445        base.join(location).map_err(|e| HttpError::InvalidUrl(e.to_string()))
446    }
447}
448
449/// True if `a` and `b` have the same scheme + host + port.
450fn same_origin(a: &Url, b: &Url) -> bool {
451    a.scheme() == b.scheme()
452        && a.host_str() == b.host_str()
453        && a.port()     == b.port()
454}
455
456/// Attach jar cookies to the request's Cookie header.
457fn attach_cookies(req: &mut Request, jar: &dyn CookieJar) {
458    let cookies = jar.cookies(&req.url);
459    if !cookies.is_empty() {
460        let pairs: Vec<String> = cookies
461            .iter()
462            .map(|c| format!("{}={}", c.name, c.value))
463            .collect();
464        req.header.set("Cookie", pairs.join("; "));
465    }
466}
467
468/// Store Set-Cookie headers from a response into the jar.
469fn store_cookies(url: &Url, header: &Header, jar: &dyn CookieJar) {
470    let cookies: Vec<Cookie> = header
471        .values("Set-Cookie")
472        .iter()
473        .filter_map(|v| {
474            let eq = v.find('=')?;
475            let name  = v[..eq].trim().to_owned();
476            let rest  = &v[eq + 1..];
477            let value = rest.split(';').next().unwrap_or("").trim().to_owned();
478            Some(Cookie::new(name, value))
479        })
480        .collect();
481    if !cookies.is_empty() {
482        jar.set_cookies(url, &cookies);
483    }
484}
485
486/// Wrap the request's context with a deadline if `timeout` is set.
487/// Returns the CancelFn (must be kept alive) and optionally a new Context.
488fn apply_timeout(req: &Request, timeout: Option<Duration>) -> (Option<CancelFn>, Option<Context>) {
489    match timeout {
490        None => (None, None),
491        Some(d) => {
492            let (ctx, cancel) = with_timeout(req.context(), d);
493            (Some(cancel), Some(ctx))
494        }
495    }
496}
497
498/// Percent-encode a form value (spaces → `+`, special chars → `%XX`).
499fn url_encode(values: &[(&str, &str)]) -> String {
500    values
501        .iter()
502        .map(|(k, v)| format!("{}={}", encode_form(k), encode_form(v)))
503        .collect::<Vec<_>>()
504        .join("&")
505}
506
507fn encode_form(s: &str) -> String {
508    let mut out = String::with_capacity(s.len());
509    for b in s.bytes() {
510        match b {
511            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9'
512            | b'-' | b'_' | b'.' | b'~' => out.push(b as char),
513            b' '                         => out.push('+'),
514            _                            => {
515                out.push('%');
516                out.push_str(&format!("{b:02X}"));
517            }
518        }
519    }
520    out
521}
522
523// ---------------------------------------------------------------------------
524// Tests
525// ---------------------------------------------------------------------------
526
527#[cfg(test)]
528mod tests {
529    use super::*;
530
531    #[test]
532    fn url_encode_basic() {
533        let pairs = [("q", "hello world"), ("lang", "rust")];
534        assert_eq!(url_encode(&pairs), "q=hello+world&lang=rust");
535    }
536
537    #[test]
538    fn url_encode_special_chars() {
539        let pairs = [("a", "b&c=d")];
540        assert_eq!(url_encode(&pairs), "a=b%26c%3Dd");
541    }
542
543    #[test]
544    fn resolve_url_absolute() {
545        let base = Url::parse("http://example.com/foo").unwrap();
546        let resolved = resolve_url(&base, "http://other.com/bar").unwrap();
547        assert_eq!(resolved.as_str(), "http://other.com/bar");
548    }
549
550    #[test]
551    fn resolve_url_relative() {
552        let base = Url::parse("http://example.com/a/b").unwrap();
553        let resolved = resolve_url(&base, "/c").unwrap();
554        assert_eq!(resolved.as_str(), "http://example.com/c");
555    }
556
557    #[test]
558    fn is_redirect_codes() {
559        for code in [301u16, 302, 303, 307, 308] {
560            assert!(is_redirect(code), "{code} should be redirect");
561        }
562        for code in [200u16, 404, 500] {
563            assert!(!is_redirect(code), "{code} should not be redirect");
564        }
565    }
566
567    #[test]
568    fn same_origin_check() {
569        let a = Url::parse("http://example.com/foo").unwrap();
570        let b = Url::parse("http://example.com/bar").unwrap();
571        let c = Url::parse("https://example.com/foo").unwrap();
572        let d = Url::parse("http://other.com/foo").unwrap();
573        assert!(same_origin(&a, &b));
574        assert!(!same_origin(&a, &c)); // different scheme
575        assert!(!same_origin(&a, &d)); // different host
576    }
577
578    #[test]
579    fn transport_pool_reuse() {
580        // Verify the pool stores and retrieves entries by key without
581        // requiring a real network connection.
582        // We can't easily test acquire() without a server, but we can
583        // verify the pool's limit enforcement via the internal state.
584        let t = Transport::new();
585        assert_eq!(t.max_idle_conns_per_host, 10);
586        // Pool starts empty.
587        assert!(t.pool.lock().unwrap().is_empty());
588    }
589
590    // client_get_end_to_end is covered by tests/server_client.rs integration
591    // tests (get_basic, multiple_sequential_requests, etc.), which carry
592    // `#[go_lib::main]` so each test body runs as the first goroutine on the
593    // shared process-wide scheduler.
594}
go_http/client.rs

go_http/
client.rs
