Skip to main content

http_net/
blocking.rs

1//! A small blocking adapter over `std::net`: the Tier-1 path.
2//!
3//! [`Server`] serves a handler on a socket with one thread per connection
4//! (bounded by [`ServerConfig::with_max_connections`]); [`Connection`] is the
5//! per-connection loop for callers who run their own accept loop or need the
6//! socket back after an upgrade; [`get`] fetches a URL. Everything here is a
7//! thin shell around [`ServerConn`] and [`ClientConn`]: all parsing, framing,
8//! and limit enforcement happens in the sans-IO core.
9//!
10//! Only plain `http://` is spoken; there is no TLS in this crate.
11
12use std::{
13    fmt, io,
14    io::{Read, Write},
15    net::{
16        IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream, ToSocketAddrs,
17    },
18    string::{String, ToString},
19    sync::{
20        Arc, Condvar, Mutex,
21        atomic::{AtomicBool, Ordering},
22    },
23    thread,
24    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
25    vec::Vec,
26};
27
28use crate::{
29    ClientConn, ClientEvent, HttpError, Limits, Method, Request, RequestHead, Response, ServerConn,
30    ServerEvent, redact::FieldValue,
31};
32
33/// Bytes a closing connection reads and discards so the peer's unread data
34/// does not turn the close into a reset that destroys the last response.
35const DRAIN_LIMIT: usize = 256 * 1024;
36/// How long one read of the lingering close may wait.
37const DRAIN_READ_TIMEOUT: Duration = Duration::from_millis(250);
38/// How long the whole lingering close may take, however the peer behaves:
39/// a peer that keeps sending must not keep the connection (and its slot).
40const DRAIN_TOTAL: Duration = Duration::from_secs(1);
41/// Responses larger than this do not keep their output buffer afterwards.
42const OUT_RETAIN: usize = 64 * 1024;
43
44/// What can go wrong in the blocking adapter.
45///
46/// # Examples
47///
48/// ```
49/// let err = http_net::blocking::get("https://example.com/").unwrap_err();
50/// assert!(matches!(err, http_net::blocking::Error::InvalidUrl(_)));
51/// ```
52#[non_exhaustive]
53#[derive(Debug)]
54pub enum Error {
55    /// The socket failed (connect, read, write, or timeout).
56    Io(io::Error),
57    /// The peer sent a message this crate refuses, or a call was out of
58    /// order.
59    Http(HttpError),
60    /// The URL is not `http://host[:port][/path][?query]`; the text says
61    /// which part is wrong.
62    InvalidUrl(&'static str),
63}
64
65impl fmt::Display for Error {
66    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
67        match self {
68            Self::Io(e) => write!(f, "I/O error: {e}"),
69            Self::Http(e) => write!(f, "HTTP error: {e}"),
70            Self::InvalidUrl(why) => write!(f, "invalid URL: {why}"),
71        }
72    }
73}
74
75impl std::error::Error for Error {
76    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
77        match self {
78            Self::Io(e) => Some(e),
79            Self::Http(e) => Some(e),
80            Self::InvalidUrl(_) => None,
81        }
82    }
83}
84
85impl From<io::Error> for Error {
86    fn from(e: io::Error) -> Self {
87        Self::Io(e)
88    }
89}
90
91impl From<HttpError> for Error {
92    fn from(e: HttpError) -> Self {
93        Self::Http(e)
94    }
95}
96
97/// Settings for [`Server`].
98///
99/// | Setting | Default |
100/// |---|---|
101/// | limits | [`Limits::new`] |
102/// | concurrent connections | 64 (the 65th is answered `503` and closed) |
103/// | read timeout (also the idle keep-alive timeout) | 10 s |
104/// | request timeout (first byte of a request to the end of its body) | 30 s |
105/// | write timeout | 10 s |
106///
107/// # Examples
108///
109/// ```
110/// use std::time::Duration;
111/// use http_net::{blocking::ServerConfig, Limits};
112///
113/// let config = ServerConfig::new()
114///     .with_limits(Limits::new().with_max_body(16 * 1024))
115///     .with_max_connections(8)
116///     .with_read_timeout(Some(Duration::from_secs(5)));
117/// assert_eq!(config.max_connections(), 8);
118/// ```
119#[derive(Clone, Copy, Debug, PartialEq, Eq)]
120pub struct ServerConfig {
121    limits: Limits,
122    max_connections: usize,
123    read_timeout: Option<Duration>,
124    request_timeout: Option<Duration>,
125    write_timeout: Option<Duration>,
126}
127
128impl Default for ServerConfig {
129    fn default() -> Self {
130        Self::new()
131    }
132}
133
134impl ServerConfig {
135    /// The defaults (see the table on [`ServerConfig`]).
136    ///
137    /// # Examples
138    ///
139    /// ```
140    /// assert_eq!(http_net::blocking::ServerConfig::new().max_connections(), 64);
141    /// ```
142    #[must_use]
143    pub const fn new() -> Self {
144        Self {
145            limits: Limits::new(),
146            max_connections: 64,
147            read_timeout: Some(Duration::from_secs(10)),
148            request_timeout: Some(Duration::from_secs(30)),
149            write_timeout: Some(Duration::from_secs(10)),
150        }
151    }
152
153    /// Sets the message limits for every connection.
154    ///
155    /// # Examples
156    ///
157    /// ```
158    /// let c = http_net::blocking::ServerConfig::new()
159    ///     .with_limits(http_net::Limits::new().with_max_headers(10));
160    /// assert_eq!(c.limits().max_headers(), 10);
161    /// ```
162    #[must_use]
163    pub const fn with_limits(mut self, limits: Limits) -> Self {
164        self.limits = limits;
165        self
166    }
167
168    /// Sets how many connections are served at once (at least 1).
169    ///
170    /// # Examples
171    ///
172    /// ```
173    /// let c = http_net::blocking::ServerConfig::new().with_max_connections(0);
174    /// assert_eq!(c.max_connections(), 1);
175    /// ```
176    #[must_use]
177    pub const fn with_max_connections(mut self, n: usize) -> Self {
178        self.max_connections = if n == 0 { 1 } else { n };
179        self
180    }
181
182    /// Sets the socket read timeout: how long a request may take to arrive,
183    /// and how long an idle keep-alive connection is held. `None` waits
184    /// forever.
185    ///
186    /// # Examples
187    ///
188    /// ```
189    /// let c = http_net::blocking::ServerConfig::new().with_read_timeout(None);
190    /// assert_eq!(c.read_timeout(), None);
191    /// ```
192    #[must_use]
193    pub const fn with_read_timeout(mut self, timeout: Option<Duration>) -> Self {
194        self.read_timeout = timeout;
195        self
196    }
197
198    /// Sets the request timeout: the most time one request may take to
199    /// arrive, from its first byte to the end of its body. The read timeout
200    /// alone cannot stop a client that sends one byte just inside it
201    /// ("slowloris"); this deadline can. A request still incomplete when it
202    /// passes is answered `408 Request Timeout` and the connection closed.
203    /// Time spent idle between keep-alive requests does not count (the read
204    /// timeout governs that). `None` disables the deadline.
205    ///
206    /// # Examples
207    ///
208    /// ```
209    /// use std::time::Duration;
210    /// let c = http_net::blocking::ServerConfig::new().with_request_timeout(Some(Duration::from_secs(10)));
211    /// assert_eq!(c.request_timeout(), Some(Duration::from_secs(10)));
212    /// ```
213    #[must_use]
214    pub const fn with_request_timeout(mut self, timeout: Option<Duration>) -> Self {
215        self.request_timeout = timeout;
216        self
217    }
218
219    /// Sets the socket write timeout. `None` waits forever.
220    ///
221    /// # Examples
222    ///
223    /// ```
224    /// use std::time::Duration;
225    /// let c = http_net::blocking::ServerConfig::new().with_write_timeout(Some(Duration::from_secs(1)));
226    /// assert_eq!(c.write_timeout(), Some(Duration::from_secs(1)));
227    /// ```
228    #[must_use]
229    pub const fn with_write_timeout(mut self, timeout: Option<Duration>) -> Self {
230        self.write_timeout = timeout;
231        self
232    }
233
234    /// The message limits.
235    ///
236    /// # Examples
237    ///
238    /// ```
239    /// assert_eq!(http_net::blocking::ServerConfig::new().limits(), http_net::Limits::new());
240    /// ```
241    #[must_use]
242    pub const fn limits(&self) -> Limits {
243        self.limits
244    }
245
246    /// The concurrent connection cap.
247    ///
248    /// # Examples
249    ///
250    /// ```
251    /// assert_eq!(http_net::blocking::ServerConfig::new().max_connections(), 64);
252    /// ```
253    #[must_use]
254    pub const fn max_connections(&self) -> usize {
255        self.max_connections
256    }
257
258    /// The read timeout.
259    ///
260    /// # Examples
261    ///
262    /// ```
263    /// use std::time::Duration;
264    /// assert_eq!(http_net::blocking::ServerConfig::new().read_timeout(), Some(Duration::from_secs(10)));
265    /// ```
266    #[must_use]
267    pub const fn read_timeout(&self) -> Option<Duration> {
268        self.read_timeout
269    }
270
271    /// The request timeout.
272    ///
273    /// # Examples
274    ///
275    /// ```
276    /// use std::time::Duration;
277    /// assert_eq!(http_net::blocking::ServerConfig::new().request_timeout(), Some(Duration::from_secs(30)));
278    /// ```
279    #[must_use]
280    pub const fn request_timeout(&self) -> Option<Duration> {
281        self.request_timeout
282    }
283
284    /// The write timeout.
285    ///
286    /// # Examples
287    ///
288    /// ```
289    /// use std::time::Duration;
290    /// assert_eq!(http_net::blocking::ServerConfig::new().write_timeout(), Some(Duration::from_secs(10)));
291    /// ```
292    #[must_use]
293    pub const fn write_timeout(&self) -> Option<Duration> {
294        self.write_timeout
295    }
296}
297
298/// A connection handed over by a `101 Switching Protocols` (or a 2xx to
299/// `CONNECT`): the socket, the bytes of the new protocol that already
300/// arrived, and the target of the request that asked for the switch.
301///
302/// # Examples
303///
304/// ```no_run
305/// use http_net::{blocking::Server, Response};
306///
307/// let server = Server::bind("127.0.0.1:0")?.on_upgrade(|up| {
308///     // Hand `up.stream` and `up.leftover` to a protocol engine.
309///     let _ = (up.stream, up.leftover, up.target);
310/// });
311/// server.serve(|req, _body| {
312///     if req.is_upgrade() {
313///         Response::switching_protocols("x")
314///     } else {
315///         Response::new(404)
316///     }
317/// })?;
318/// # Ok::<(), std::io::Error>(())
319/// ```
320#[derive(Debug)]
321pub struct Upgraded {
322    /// The socket, now speaking the new protocol.
323    pub stream: TcpStream,
324    /// Bytes of the new protocol received together with the request.
325    pub leftover: Vec<u8>,
326    /// The target of the request that asked for the switch.
327    pub target: String,
328}
329
330type UpgradeHook = Arc<dyn Fn(Upgraded) + Send + Sync>;
331
332/// A blocking HTTP/1.1 server: one thread per connection, bounded.
333///
334/// # Examples
335///
336/// ```
337/// use http_net::{blocking::{self, Server}, Response};
338///
339/// let server = Server::bind("127.0.0.1:0")?;
340/// let addr = server.local_addr();
341/// let stop = server.shutdown_handle();
342/// let thread = std::thread::spawn(move || {
343///     server.serve(|req, _body| Response::text(200, &format!("you asked for {}", req.path())))
344/// });
345///
346/// let resp = blocking::get(&format!("http://{addr}/hello"))?;
347/// assert_eq!(resp.status(), 200);
348/// assert_eq!(resp.text(), Some("you asked for /hello"));
349///
350/// stop.shutdown();
351/// thread.join().unwrap()?;
352/// # Ok::<(), Box<dyn std::error::Error>>(())
353/// ```
354pub struct Server {
355    listener: TcpListener,
356    local: SocketAddr,
357    config: ServerConfig,
358    stop: Arc<AtomicBool>,
359    upgrade: Option<UpgradeHook>,
360}
361
362impl fmt::Debug for Server {
363    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
364        f.debug_struct("Server")
365            .field("local_addr", &self.local)
366            .field("config", &self.config)
367            .finish()
368    }
369}
370
371/// Stops a running [`Server::serve`] from another thread.
372///
373/// # Examples
374///
375/// ```
376/// let server = http_net::blocking::Server::bind("127.0.0.1:0")?;
377/// let stop = server.shutdown_handle();
378/// let t = std::thread::spawn(move || server.serve(|_, _| http_net::Response::new(204)));
379/// stop.shutdown();
380/// t.join().unwrap()?;
381/// # Ok::<(), std::io::Error>(())
382/// ```
383#[derive(Clone, Debug)]
384pub struct ShutdownHandle {
385    stop: Arc<AtomicBool>,
386    wake: SocketAddr,
387}
388
389impl ShutdownHandle {
390    /// Asks the server to stop: it accepts no new connections, lets each
391    /// open connection finish its current exchange, waits for them, and
392    /// returns from `serve`.
393    ///
394    /// # Examples
395    ///
396    /// See [`ShutdownHandle`].
397    pub fn shutdown(&self) {
398        self.stop.store(true, Ordering::SeqCst);
399        // `accept` blocks; a throwaway connection wakes it so it sees the
400        // flag. If the connect fails the listener is already gone, which is
401        // the outcome being asked for, so the error carries no information.
402        if let Ok(stream) = TcpStream::connect_timeout(&self.wake, Duration::from_secs(1)) {
403            drop(stream);
404        }
405    }
406}
407
408/// Counts live connection threads so shutdown can wait for them.
409#[derive(Default)]
410struct Active {
411    count: Mutex<usize>,
412    idle: Condvar,
413}
414
415impl Active {
416    fn try_enter(&self, max: usize) -> bool {
417        let mut n = self
418            .count
419            .lock()
420            .unwrap_or_else(std::sync::PoisonError::into_inner);
421        if *n >= max {
422            return false;
423        }
424        *n += 1;
425        true
426    }
427
428    fn leave(&self) {
429        let mut n = self
430            .count
431            .lock()
432            .unwrap_or_else(std::sync::PoisonError::into_inner);
433        *n = n.saturating_sub(1);
434        if *n == 0 {
435            self.idle.notify_all();
436        }
437    }
438
439    fn wait_idle(&self) {
440        let mut n = self
441            .count
442            .lock()
443            .unwrap_or_else(std::sync::PoisonError::into_inner);
444        while *n > 0 {
445            n = self
446                .idle
447                .wait(n)
448                .unwrap_or_else(std::sync::PoisonError::into_inner);
449        }
450    }
451}
452
453/// Leaves the active set even if the handler panics.
454struct ActiveGuard(Arc<Active>);
455
456impl Drop for ActiveGuard {
457    fn drop(&mut self) {
458        self.0.leave();
459    }
460}
461
462impl Server {
463    /// Binds a listening socket. Use `127.0.0.1:0` for an ephemeral
464    /// localhost port and read it back with [`local_addr`](Self::local_addr).
465    ///
466    /// # Errors
467    ///
468    /// The bind error from the operating system.
469    ///
470    /// # Examples
471    ///
472    /// ```
473    /// let server = http_net::blocking::Server::bind("127.0.0.1:0")?;
474    /// assert!(server.local_addr().ip().is_loopback());
475    /// # Ok::<(), std::io::Error>(())
476    /// ```
477    pub fn bind(addr: impl ToSocketAddrs) -> io::Result<Self> {
478        let listener = TcpListener::bind(addr)?;
479        let local = listener.local_addr()?;
480        Ok(Self {
481            listener,
482            local,
483            config: ServerConfig::new(),
484            stop: Arc::new(AtomicBool::new(false)),
485            upgrade: None,
486        })
487    }
488
489    /// Replaces the configuration.
490    ///
491    /// # Examples
492    ///
493    /// ```
494    /// use http_net::blocking::{Server, ServerConfig};
495    /// let s = Server::bind("127.0.0.1:0")?.with_config(ServerConfig::new().with_max_connections(4));
496    /// # Ok::<(), std::io::Error>(())
497    /// ```
498    #[must_use]
499    pub fn with_config(mut self, config: ServerConfig) -> Self {
500        self.config = config;
501        self
502    }
503
504    /// Sets what happens to a connection after the handler answers with
505    /// `101 Switching Protocols`: `hook` receives the socket on the
506    /// connection's thread. Without a hook such connections are closed.
507    ///
508    /// # Examples
509    ///
510    /// See [`Upgraded`].
511    #[must_use]
512    pub fn on_upgrade(mut self, hook: impl Fn(Upgraded) + Send + Sync + 'static) -> Self {
513        self.upgrade = Some(Arc::new(hook));
514        self
515    }
516
517    /// The bound address (with the real port when bound to port 0).
518    ///
519    /// # Examples
520    ///
521    /// ```
522    /// let server = http_net::blocking::Server::bind("127.0.0.1:0")?;
523    /// assert_ne!(server.local_addr().port(), 0);
524    /// # Ok::<(), std::io::Error>(())
525    /// ```
526    #[must_use]
527    pub fn local_addr(&self) -> SocketAddr {
528        self.local
529    }
530
531    /// A handle that stops [`serve`](Self::serve) from another thread.
532    ///
533    /// # Examples
534    ///
535    /// See [`ShutdownHandle`].
536    #[must_use]
537    pub fn shutdown_handle(&self) -> ShutdownHandle {
538        let mut wake = self.local;
539        if wake.ip().is_unspecified() {
540            wake.set_ip(match wake.ip() {
541                IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST),
542                IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST),
543            });
544        }
545        ShutdownHandle {
546            stop: Arc::clone(&self.stop),
547            wake,
548        }
549    }
550
551    /// Serves `handler` until [`ShutdownHandle::shutdown`] is called.
552    ///
553    /// The handler gets each request's head and its complete body (bounded
554    /// by [`Limits::max_body`]) and returns the response. Requests that fail
555    /// to parse are answered with their [`HttpError::status`] and the
556    /// connection is closed; the handler never sees them. Every response
557    /// gets a `date` header unless the handler set one.
558    ///
559    /// # Errors
560    ///
561    /// Only if accepting connections fails persistently; transient accept
562    /// errors are retried.
563    ///
564    /// # Examples
565    ///
566    /// See [`Server`].
567    pub fn serve<H>(self, handler: H) -> io::Result<()>
568    where
569        H: Fn(&RequestHead<'_>, &[u8]) -> Response + Send + Sync + 'static,
570    {
571        let handler = Arc::new(handler);
572        let active = Arc::new(Active::default());
573        let mut consecutive_errors = 0u32;
574        for incoming in self.listener.incoming() {
575            if self.stop.load(Ordering::SeqCst) {
576                break;
577            }
578            let stream = match incoming {
579                Ok(stream) => {
580                    consecutive_errors = 0;
581                    stream
582                }
583                Err(e) => {
584                    // Aborted handshakes and descriptor exhaustion are
585                    // transient; back off briefly instead of spinning, and
586                    // give up only if accept never recovers.
587                    consecutive_errors += 1;
588                    if consecutive_errors > 1000 {
589                        return Err(e);
590                    }
591                    thread::sleep(Duration::from_millis(10));
592                    continue;
593                }
594            };
595            if !active.try_enter(self.config.max_connections) {
596                refuse_busy(stream, &self.config);
597                continue;
598            }
599            let guard = ActiveGuard(Arc::clone(&active));
600            let handler = Arc::clone(&handler);
601            let upgrade = self.upgrade.clone();
602            let stop = Arc::clone(&self.stop);
603            let config = self.config;
604            let spawned = thread::Builder::new()
605                .name("http-net-conn".into())
606                .spawn(move || {
607                    let _guard = guard;
608                    serve_stream(stream, &config, &*handler, upgrade.as_deref(), &stop);
609                });
610            // A failed spawn drops the closure, its guard, and the stream:
611            // the connection is closed and the slot released.
612            drop(spawned);
613        }
614        active.wait_idle();
615        Ok(())
616    }
617}
618
619fn refuse_busy(mut stream: TcpStream, config: &ServerConfig) {
620    const BUSY: &[u8] =
621        b"HTTP/1.1 503 Service Unavailable\r\ncontent-length: 0\r\nconnection: close\r\n\r\n";
622    // Best effort on a connection being turned away: a failure here only
623    // means the client already left. This runs on the accept thread, so it
624    // must never wait on the peer: the 503 fits in an empty send buffer,
625    // and only what has already arrived is drained (without blocking) so
626    // that it does not turn the close into a reset.
627    if stream.set_write_timeout(config.write_timeout).is_err()
628        || stream.write_all(BUSY).is_err()
629        || stream.shutdown(Shutdown::Write).is_err()
630        || stream.set_nonblocking(true).is_err()
631    {
632        return;
633    }
634    let mut scratch = [0u8; 4096];
635    let mut drained = 0;
636    while drained < DRAIN_LIMIT {
637        match stream.read(&mut scratch) {
638            Ok(n) if n > 0 => drained += n,
639            _ => break,
640        }
641    }
642}
643
644fn serve_stream(
645    stream: TcpStream,
646    config: &ServerConfig,
647    handler: &(dyn Fn(&RequestHead<'_>, &[u8]) -> Response + Send + Sync),
648    upgrade: Option<&(dyn Fn(Upgraded) + Send + Sync)>,
649    stop: &AtomicBool,
650) {
651    let configured = stream.set_read_timeout(config.read_timeout).is_ok()
652        && stream.set_write_timeout(config.write_timeout).is_ok()
653        && stream.set_nodelay(true).is_ok();
654    if !configured {
655        return;
656    }
657    let mut conn =
658        Connection::new(stream, config.limits).with_request_timeout(config.request_timeout);
659    while !stop.load(Ordering::SeqCst) {
660        // The handler runs inside the first match; errors (which borrow
661        // nothing) are handled in the second, where `conn` is free again.
662        let outcome = match conn.next_request() {
663            Ok(Some(incoming)) => {
664                let head = incoming.head();
665                Ok((
666                    handler(&head, incoming.body()),
667                    head.method() == Method::Head,
668                ))
669            }
670            Ok(None) => break,
671            Err(e) => Err(e),
672        };
673        let (response, is_head) = match outcome {
674            Ok(answer) => answer,
675            Err(Error::Http(e)) => {
676                if conn.respond(&Response::from_error(&e)).is_err() {
677                    return;
678                }
679                break;
680            }
681            Err(Error::Io(e)) if is_timeout(&e) && conn.request_in_progress() => {
682                // A request that did not arrive in time: say so, then
683                // close (the rest of it can no longer be framed).
684                if conn.send_request_timeout().is_ok() {
685                    conn.close();
686                }
687                return;
688            }
689            Err(_) => return,
690        };
691        // `Server` writes full bodies only: the handler has no way to send
692        // a streamed one, so it would leave the client waiting. A `HEAD`
693        // answer is the exception: it announces a length or chunked
694        // framing but carries no body, so it is complete once written.
695        let streamed = response.body_bytes().is_none();
696        let response = if streamed && !is_head {
697            Response::text(
698                500,
699                "handler error: Server cannot send a streamed response; \
700                 use Connection for streaming",
701            )
702            .close()
703        } else {
704            response
705        };
706        match conn.respond(&response) {
707            Ok(()) if streamed && is_head => {
708                if conn.end_body().is_err() {
709                    return;
710                }
711            }
712            Ok(()) => {}
713            // Refused before anything was written (a builder error, or a
714            // 101 to a request that did not ask): answer 500 instead.
715            Err(Error::Http(e)) => {
716                let message = format!("handler error: invalid response: {e}");
717                if conn
718                    .respond(&Response::text(500, &message).close())
719                    .is_err()
720                {
721                    return;
722                }
723            }
724            Err(_) => return,
725        }
726        if conn.is_upgraded() {
727            match (conn.into_upgraded(), upgrade) {
728                (Some(up), Some(hook)) => hook(up),
729                (Some(up), None) => drain_and_close(up.stream),
730                (None, _) => {}
731            }
732            return;
733        }
734    }
735    conn.close();
736}
737
738fn drain_and_close(mut stream: TcpStream) {
739    // Lingering close (RFC 9112 section 9.6): stop sending, then read what
740    // the peer is still sending so the close does not become a reset that
741    // discards the response before the peer reads it.
742    // The drain is bounded in bytes and in total time, not only per read:
743    // a peer that keeps trickling data must not hold the thread.
744    if stream.flush().is_err() || stream.shutdown(Shutdown::Write).is_err() {
745        return;
746    }
747    let deadline = Instant::now() + DRAIN_TOTAL;
748    let mut scratch = [0u8; 4096];
749    let mut drained = 0;
750    while drained < DRAIN_LIMIT {
751        let left = deadline.saturating_duration_since(Instant::now());
752        if left.is_zero()
753            || stream
754                .set_read_timeout(Some(left.min(DRAIN_READ_TIMEOUT)))
755                .is_err()
756        {
757            break;
758        }
759        match stream.read(&mut scratch) {
760            Ok(0) | Err(_) => break,
761            Ok(n) => drained += n,
762        }
763    }
764}
765
766fn is_timeout(e: &io::Error) -> bool {
767    // Unix reports an expired socket timeout as WouldBlock, Windows as
768    // TimedOut.
769    matches!(
770        e.kind(),
771        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
772    )
773}
774
775/// One request read by [`Connection::next_request`]: its head and its
776/// complete body.
777///
778/// # Examples
779///
780/// See [`Connection`].
781#[derive(Debug)]
782pub struct Incoming<'a> {
783    head: RequestHead<'a>,
784    body: &'a [u8],
785}
786
787impl<'a> Incoming<'a> {
788    /// The request head.
789    ///
790    /// # Examples
791    ///
792    /// See [`Connection`].
793    #[must_use]
794    pub fn head(&self) -> RequestHead<'a> {
795        self.head
796    }
797
798    /// The complete request body (empty when there is none).
799    ///
800    /// # Examples
801    ///
802    /// See [`Connection`].
803    #[must_use]
804    pub fn body(&self) -> &'a [u8] {
805        self.body
806    }
807}
808
809/// A blocking server connection: [`ServerConn`] driven over a `TcpStream`.
810///
811/// Use it for your own accept loop, or to keep the socket after an
812/// upgrade with [`into_upgraded`](Self::into_upgraded).
813///
814/// # Examples
815///
816/// ```
817/// use std::net::{TcpListener, TcpStream};
818/// use std::io::{Read, Write};
819/// use http_net::{blocking::Connection, Limits, Response};
820///
821/// let listener = TcpListener::bind("127.0.0.1:0")?;
822/// let addr = listener.local_addr()?;
823/// let client = std::thread::spawn(move || -> std::io::Result<Vec<u8>> {
824///     let mut s = TcpStream::connect(addr)?;
825///     s.write_all(b"POST /sum HTTP/1.1\r\nHost: x\r\nContent-Length: 3\r\nConnection: close\r\n\r\n1 2")?;
826///     let mut reply = Vec::new();
827///     s.read_to_end(&mut reply)?;
828///     Ok(reply)
829/// });
830///
831/// let (stream, _) = listener.accept()?;
832/// let mut conn = Connection::new(stream, Limits::new());
833/// while let Some(req) = conn.next_request()? {
834///     let sum: u32 = std::str::from_utf8(req.body())?.split(' ').filter_map(|n| n.parse::<u32>().ok()).sum();
835///     let response = Response::text(200, &sum.to_string());
836///     conn.respond(&response)?;
837/// }
838/// conn.close();
839/// let reply = client.join().unwrap()?;
840/// assert!(reply.ends_with(b"\r\n\r\n3"));
841/// # Ok::<(), Box<dyn std::error::Error>>(())
842/// ```
843#[derive(Debug)]
844pub struct Connection {
845    stream: TcpStream,
846    conn: ServerConn,
847    body: Vec<u8>,
848    out: Vec<u8>,
849    /// The socket's own read timeout, restored once a request deadline no
850    /// longer applies.
851    read_timeout: Option<Duration>,
852    request_timeout: Option<Duration>,
853    /// When the request being read must be complete.
854    deadline: Option<Instant>,
855    /// Whether the socket's read timeout is currently shortened to the
856    /// deadline.
857    shortened: bool,
858}
859
860impl Connection {
861    /// Wraps an accepted socket. Set its timeouts first if you want them:
862    /// the read timeout set on `stream` is the idle timeout, and
863    /// [`with_request_timeout`](Self::with_request_timeout) adds a deadline
864    /// for each request as a whole.
865    ///
866    /// # Examples
867    ///
868    /// See [`Connection`].
869    #[must_use]
870    pub fn new(stream: TcpStream, limits: Limits) -> Self {
871        let read_timeout = stream.read_timeout().ok().flatten();
872        Self {
873            stream,
874            conn: ServerConn::with_limits(limits),
875            body: Vec::new(),
876            out: Vec::new(),
877            read_timeout,
878            request_timeout: None,
879            deadline: None,
880            shortened: false,
881        }
882    }
883
884    /// Sets a deadline for each request: from the moment its first byte is
885    /// seen to the end of its body (see
886    /// [`ServerConfig::with_request_timeout`]). When it passes,
887    /// [`next_request`](Self::next_request) fails with an [`Error::Io`] of
888    /// kind `TimedOut`. `None` (the default for a bare `Connection`) leaves
889    /// only the socket's own read timeout.
890    ///
891    /// # Examples
892    ///
893    /// ```no_run
894    /// use std::time::Duration;
895    /// # fn demo(stream: std::net::TcpStream) {
896    /// let conn = http_net::blocking::Connection::new(stream, http_net::Limits::new())
897    ///     .with_request_timeout(Some(Duration::from_secs(10)));
898    /// # let _ = conn;
899    /// # }
900    /// ```
901    #[must_use]
902    pub fn with_request_timeout(mut self, timeout: Option<Duration>) -> Self {
903        self.request_timeout = timeout;
904        self
905    }
906
907    /// Whether part of a request has arrived and the rest is outstanding.
908    fn request_in_progress(&self) -> bool {
909        self.conn.request().is_some() || !self.conn.remaining_input().is_empty()
910    }
911
912    /// Answers a request that did not arrive in time with `408`, unless a
913    /// response is already on its way.
914    fn send_request_timeout(&mut self) -> Result<(), Error> {
915        if self.conn.response_started() {
916            return Err(Error::Http(HttpError::InvalidState(
917                "a response was already started",
918            )));
919        }
920        const BODY: &[u8] = b"request timed out";
921        self.out.clear();
922        self.out.extend_from_slice(
923            b"HTTP/1.1 408 Request Timeout\r\ncontent-type: text/plain; charset=utf-8\r\n\
924              content-length: ",
925        );
926        self.out
927            .extend_from_slice(BODY.len().to_string().as_bytes());
928        self.out
929            .extend_from_slice(b"\r\nconnection: close\r\ndate: ");
930        self.out.extend_from_slice(&http_date(SystemTime::now()));
931        self.out.extend_from_slice(b"\r\n\r\n");
932        self.out.extend_from_slice(BODY);
933        self.flush_out()
934    }
935
936    /// Reads into the parser's buffer, under the request deadline if one
937    /// is running.
938    fn fill(&mut self) -> Result<(), Error> {
939        if let Some(limit) = self.request_timeout {
940            if self.deadline.is_none() && self.request_in_progress() {
941                self.deadline = Some(Instant::now() + limit);
942            }
943        }
944        match self.deadline {
945            Some(deadline) => {
946                let left = deadline.saturating_duration_since(Instant::now());
947                if left.is_zero() {
948                    return Err(Error::Io(io::Error::new(
949                        io::ErrorKind::TimedOut,
950                        "the request did not arrive within the request timeout",
951                    )));
952                }
953                let wait = self.read_timeout.map_or(left, |t| t.min(left));
954                self.stream.set_read_timeout(Some(wait))?;
955                self.shortened = true;
956            }
957            None if self.shortened => {
958                self.stream.set_read_timeout(self.read_timeout)?;
959                self.shortened = false;
960            }
961            None => {}
962        }
963        let spare = self.conn.read_buf();
964        if spare.is_empty() {
965            return Err(Error::Http(HttpError::InvalidState(
966                "input buffer full while the parser needs input",
967            )));
968        }
969        let n = loop {
970            match self.stream.read(spare) {
971                Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
972                other => break other?,
973            }
974        };
975        if n == 0 {
976            self.conn.feed_eof();
977        } else {
978            self.conn.commit(n);
979        }
980        Ok(())
981    }
982
983    /// Reads the next complete request. `Ok(None)` means the connection is
984    /// finished (the client closed between requests, the previous exchange
985    /// ended with `connection: close`, or the connection was upgraded).
986    ///
987    /// A request that carries `Expect: 100-continue` is answered with
988    /// `100 Continue` before its body is read.
989    ///
990    /// # Errors
991    ///
992    /// [`Error::Http`] for a request this crate refuses (answer it with its
993    /// [`HttpError::status`] and close), [`Error::Io`] for socket failures
994    /// and timeouts (including the request deadline, as `TimedOut`), and
995    /// [`Error::Http`] with [`HttpError::InvalidState`] if the previous
996    /// request has not been answered.
997    ///
998    /// # Examples
999    ///
1000    /// See [`Connection`].
1001    pub fn next_request(&mut self) -> Result<Option<Incoming<'_>>, Error> {
1002        self.body.clear();
1003        self.deadline = None;
1004        loop {
1005            match self.conn.poll()? {
1006                ServerEvent::Request(head) => {
1007                    // The whole body is read before the caller sees the
1008                    // request, so the expectation is answered here (RFC 9110
1009                    // section 10.1.1). A body over the limit was already
1010                    // refused from its Content-Length.
1011                    if head.expects_continue() {
1012                        self.out.clear();
1013                        self.conn.send_continue(&mut self.out)?;
1014                        self.flush_out()?;
1015                    }
1016                }
1017                ServerEvent::Body(chunk) => self.body.extend_from_slice(chunk),
1018                ServerEvent::RequestEnd => break,
1019                ServerEvent::NeedInput => self.fill()?,
1020                ServerEvent::AwaitingResponse => {
1021                    return Err(Error::Http(HttpError::InvalidState(
1022                        "respond to the previous request first",
1023                    )));
1024                }
1025                _ => return Ok(None),
1026            }
1027        }
1028        match self.conn.request() {
1029            Some(head) => Ok(Some(Incoming {
1030                head,
1031                body: &self.body,
1032            })),
1033            None => Ok(None),
1034        }
1035    }
1036
1037    /// Writes `response` to the socket. A streamed response continues with
1038    /// [`send_body`](Self::send_body) and [`end_body`](Self::end_body).
1039    ///
1040    /// # Errors
1041    ///
1042    /// [`Error::Http`] when the response is invalid or out of order (see
1043    /// [`ServerConn::respond`]); [`Error::Io`] when the write fails.
1044    ///
1045    /// # Examples
1046    ///
1047    /// See [`Connection`].
1048    pub fn respond(&mut self, response: &Response) -> Result<(), Error> {
1049        self.out.clear();
1050        let date = http_date(SystemTime::now());
1051        self.conn
1052            .respond_dated(response, Some(&date), &mut self.out)?;
1053        self.flush_out()
1054    }
1055
1056    /// Writes body bytes of a streamed response.
1057    ///
1058    /// # Errors
1059    ///
1060    /// As [`ServerConn::send_body`], plus [`Error::Io`].
1061    ///
1062    /// # Examples
1063    ///
1064    /// ```no_run
1065    /// # fn demo(conn: &mut http_net::blocking::Connection) -> Result<(), http_net::blocking::Error> {
1066    /// conn.respond(&http_net::Response::new(200).chunked())?;
1067    /// conn.send_body(b"first")?;
1068    /// conn.end_body()?;
1069    /// # Ok(()) }
1070    /// ```
1071    pub fn send_body(&mut self, data: &[u8]) -> Result<(), Error> {
1072        self.out.clear();
1073        self.conn.send_body(data, &mut self.out)?;
1074        self.flush_out()
1075    }
1076
1077    /// Ends a streamed response.
1078    ///
1079    /// # Errors
1080    ///
1081    /// As [`ServerConn::end_body`], plus [`Error::Io`].
1082    ///
1083    /// # Examples
1084    ///
1085    /// See [`send_body`](Self::send_body).
1086    pub fn end_body(&mut self) -> Result<(), Error> {
1087        self.out.clear();
1088        self.conn.end_body(&mut self.out)?;
1089        self.flush_out()
1090    }
1091
1092    fn flush_out(&mut self) -> Result<(), Error> {
1093        self.stream.write_all(&self.out)?;
1094        if self.out.capacity() > OUT_RETAIN {
1095            self.out = Vec::new();
1096        }
1097        Ok(())
1098    }
1099
1100    /// Whether a `101` (or a 2xx to `CONNECT`) was sent.
1101    ///
1102    /// # Examples
1103    ///
1104    /// See [`into_upgraded`](Self::into_upgraded).
1105    #[must_use]
1106    pub fn is_upgraded(&self) -> bool {
1107        self.conn.is_upgraded()
1108    }
1109
1110    /// The socket and the bytes of the new protocol, after an upgrade;
1111    /// `None` if the connection was not upgraded. The socket's read timeout
1112    /// is the one it had when the `Connection` was created (a request
1113    /// deadline may have shortened it while the request was read).
1114    ///
1115    /// # Examples
1116    ///
1117    /// ```no_run
1118    /// # fn demo(mut conn: http_net::blocking::Connection) {
1119    /// if conn.is_upgraded() {
1120    ///     let up = conn.into_upgraded().unwrap();
1121    ///     // speak the new protocol on up.stream, starting with up.leftover
1122    /// #   let _ = up;
1123    /// }
1124    /// # }
1125    /// ```
1126    #[must_use]
1127    pub fn into_upgraded(self) -> Option<Upgraded> {
1128        if !self.is_upgraded() {
1129            return None;
1130        }
1131        let target = self
1132            .conn
1133            .request()
1134            .map(|r| r.target().to_string())
1135            .unwrap_or_default();
1136        if self.shortened {
1137            // Best effort: the new protocol must not inherit what was left
1138            // of the request deadline. Setting a timeout this socket has
1139            // already accepted does not fail in practice, and the hand-off
1140            // is more useful than refusing it over one.
1141            let _ = self.stream.set_read_timeout(self.read_timeout);
1142        }
1143        Some(Upgraded {
1144            leftover: self.conn.remaining_input().to_vec(),
1145            target,
1146            stream: self.stream,
1147        })
1148    }
1149
1150    /// Closes the connection gracefully: stops sending, briefly drains what
1151    /// the client is still sending, and drops the socket.
1152    ///
1153    /// # Examples
1154    ///
1155    /// See [`Connection`].
1156    pub fn close(self) {
1157        drain_and_close(self.stream);
1158    }
1159}
1160
1161/// A response fetched by [`get`] or [`Client::get`], fully read.
1162///
1163/// # Examples
1164///
1165/// See [`Server`].
1166#[derive(Clone, PartialEq, Eq)]
1167pub struct ClientResponse {
1168    status: u16,
1169    headers: Vec<(String, Vec<u8>)>,
1170    body: Vec<u8>,
1171}
1172
1173/// Shows status, fields, and body size; the values of credential fields
1174/// (`set-cookie` above all) are replaced by `<redacted N bytes>`.
1175impl fmt::Debug for ClientResponse {
1176    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1177        struct Fields<'a>(&'a [(String, Vec<u8>)]);
1178        impl fmt::Debug for Fields<'_> {
1179            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1180                f.debug_list()
1181                    .entries(self.0.iter().map(|(name, value)| {
1182                        (
1183                            name.as_str(),
1184                            FieldValue {
1185                                name: name.as_bytes(),
1186                                value,
1187                            },
1188                        )
1189                    }))
1190                    .finish()
1191            }
1192        }
1193        f.debug_struct("ClientResponse")
1194            .field("status", &self.status)
1195            .field("headers", &Fields(&self.headers))
1196            .field("body", &format_args!("{} bytes", self.body.len()))
1197            .finish()
1198    }
1199}
1200
1201impl ClientResponse {
1202    /// The status code.
1203    ///
1204    /// # Examples
1205    ///
1206    /// See [`Server`].
1207    #[must_use]
1208    pub fn status(&self) -> u16 {
1209        self.status
1210    }
1211
1212    /// The fields in arrival order.
1213    ///
1214    /// # Examples
1215    ///
1216    /// ```no_run
1217    /// let resp = http_net::blocking::get("http://127.0.0.1:8080/")?;
1218    /// for (name, value) in resp.headers() {
1219    ///     let _ = (name, value);
1220    /// }
1221    /// # Ok::<(), http_net::blocking::Error>(())
1222    /// ```
1223    pub fn headers(&self) -> impl Iterator<Item = (&str, &[u8])> {
1224        self.headers.iter().map(|(n, v)| (n.as_str(), v.as_slice()))
1225    }
1226
1227    /// The first field named `name` (ASCII case-insensitive).
1228    ///
1229    /// # Examples
1230    ///
1231    /// ```no_run
1232    /// let resp = http_net::blocking::get("http://127.0.0.1:8080/")?;
1233    /// let _ = resp.header("content-type");
1234    /// # Ok::<(), http_net::blocking::Error>(())
1235    /// ```
1236    #[must_use]
1237    pub fn header(&self, name: &str) -> Option<&[u8]> {
1238        self.headers
1239            .iter()
1240            .find(|(n, _)| n.eq_ignore_ascii_case(name))
1241            .map(|(_, v)| v.as_slice())
1242    }
1243
1244    /// The body.
1245    ///
1246    /// # Examples
1247    ///
1248    /// See [`Server`].
1249    #[must_use]
1250    pub fn body(&self) -> &[u8] {
1251        &self.body
1252    }
1253
1254    /// The body, if it is UTF-8.
1255    ///
1256    /// # Examples
1257    ///
1258    /// See [`Server`].
1259    #[must_use]
1260    pub fn text(&self) -> Option<&str> {
1261        core::str::from_utf8(&self.body).ok()
1262    }
1263
1264    /// Takes the body.
1265    ///
1266    /// # Examples
1267    ///
1268    /// ```no_run
1269    /// let bytes = http_net::blocking::get("http://127.0.0.1:8080/")?.into_body();
1270    /// # let _ = bytes;
1271    /// # Ok::<(), http_net::blocking::Error>(())
1272    /// ```
1273    #[must_use]
1274    pub fn into_body(self) -> Vec<u8> {
1275        self.body
1276    }
1277}
1278
1279/// Fetches `url` with `GET` using the default [`Client`].
1280///
1281/// # Errors
1282///
1283/// [`Error::InvalidUrl`] for anything but `http://host[:port][/path]`,
1284/// [`Error::Io`] for connection failures and timeouts, [`Error::Http`] for
1285/// a response this crate refuses (including one over the body limit).
1286///
1287/// # Examples
1288///
1289/// See [`Server`].
1290pub fn get(url: &str) -> Result<ClientResponse, Error> {
1291    Client::new().get(url)
1292}
1293
1294/// A blocking HTTP/1.1 client: one connection per request.
1295///
1296/// | Setting | Default |
1297/// |---|---|
1298/// | limits | [`Limits::new`] with a 16 MiB body |
1299/// | connect, read, and write timeout | 30 s each |
1300///
1301/// # Examples
1302///
1303/// ```no_run
1304/// use std::time::Duration;
1305/// use http_net::{blocking::Client, Limits};
1306///
1307/// let client = Client::new()
1308///     .with_limits(Limits::new().with_max_body(64 << 20))
1309///     .with_timeout(Duration::from_secs(5));
1310/// let resp = client.get("http://127.0.0.1:8080/big")?;
1311/// # let _ = resp;
1312/// # Ok::<(), http_net::blocking::Error>(())
1313/// ```
1314#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1315pub struct Client {
1316    limits: Limits,
1317    timeout: Duration,
1318}
1319
1320impl Default for Client {
1321    fn default() -> Self {
1322        Self::new()
1323    }
1324}
1325
1326impl Client {
1327    /// The defaults (see the table on [`Client`]).
1328    ///
1329    /// # Examples
1330    ///
1331    /// ```
1332    /// let _ = http_net::blocking::Client::new();
1333    /// ```
1334    #[must_use]
1335    pub const fn new() -> Self {
1336        Self {
1337            limits: Limits::new().with_max_body(16 * 1024 * 1024),
1338            timeout: Duration::from_secs(30),
1339        }
1340    }
1341
1342    /// Sets the response limits.
1343    ///
1344    /// # Examples
1345    ///
1346    /// ```
1347    /// let _ = http_net::blocking::Client::new().with_limits(http_net::Limits::new());
1348    /// ```
1349    #[must_use]
1350    pub const fn with_limits(mut self, limits: Limits) -> Self {
1351        self.limits = limits;
1352        self
1353    }
1354
1355    /// Sets the connect, read, and write timeout (zero is raised to 1 ms).
1356    ///
1357    /// # Examples
1358    ///
1359    /// ```
1360    /// let _ = http_net::blocking::Client::new().with_timeout(std::time::Duration::from_secs(2));
1361    /// ```
1362    #[must_use]
1363    pub const fn with_timeout(mut self, timeout: Duration) -> Self {
1364        self.timeout = if timeout.is_zero() {
1365            Duration::from_millis(1)
1366        } else {
1367            timeout
1368        };
1369        self
1370    }
1371
1372    /// Fetches `url` with `GET`.
1373    ///
1374    /// # Errors
1375    ///
1376    /// As [`get`].
1377    ///
1378    /// # Examples
1379    ///
1380    /// See [`Client`].
1381    pub fn get(&self, url: &str) -> Result<ClientResponse, Error> {
1382        let url = Url::parse(url)?;
1383        let request = Request::get(url.target)
1384            .header("host", url.authority)
1385            .close();
1386        let mut stream = self.connect(&url)?;
1387        let mut conn = ClientConn::with_limits(self.limits);
1388        let mut out = Vec::new();
1389        conn.send_request(&request, &mut out)?;
1390        stream.write_all(&out)?;
1391        read_response(&mut stream, &mut conn)
1392    }
1393
1394    fn connect(&self, url: &Url<'_>) -> Result<TcpStream, Error> {
1395        let mut last = None;
1396        for addr in (url.host, url.port).to_socket_addrs()? {
1397            match TcpStream::connect_timeout(&addr, self.timeout) {
1398                Ok(stream) => {
1399                    stream.set_read_timeout(Some(self.timeout))?;
1400                    stream.set_write_timeout(Some(self.timeout))?;
1401                    stream.set_nodelay(true)?;
1402                    return Ok(stream);
1403                }
1404                Err(e) => last = Some(e),
1405            }
1406        }
1407        Err(Error::Io(last.unwrap_or_else(|| {
1408            io::Error::new(io::ErrorKind::NotFound, "the host resolved to no address")
1409        })))
1410    }
1411}
1412
1413fn read_response(stream: &mut TcpStream, conn: &mut ClientConn) -> Result<ClientResponse, Error> {
1414    let mut response = ClientResponse {
1415        status: 0,
1416        headers: Vec::new(),
1417        body: Vec::new(),
1418    };
1419    loop {
1420        match conn.poll()? {
1421            ClientEvent::Response(head) => {
1422                response.status = head.status();
1423                response.headers = head
1424                    .headers()
1425                    .map(|(n, v)| (n.to_string(), v.to_vec()))
1426                    .collect();
1427            }
1428            ClientEvent::Body(chunk) => response.body.extend_from_slice(chunk),
1429            ClientEvent::ResponseEnd => return Ok(response),
1430            ClientEvent::NeedInput => {
1431                let spare = conn.read_buf();
1432                if spare.is_empty() {
1433                    return Err(Error::Http(HttpError::InvalidState(
1434                        "input buffer full while the parser needs input",
1435                    )));
1436                }
1437                let n = loop {
1438                    match stream.read(spare) {
1439                        Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
1440                        other => break other?,
1441                    }
1442                };
1443                if n == 0 {
1444                    conn.feed_eof();
1445                } else {
1446                    conn.commit(n);
1447                }
1448            }
1449            _ => {
1450                return Err(Error::Http(HttpError::InvalidState(
1451                    "the connection ended without a complete response",
1452                )));
1453            }
1454        }
1455    }
1456}
1457
1458/// The parts of an `http://` URL needed to send a request.
1459#[derive(Debug, PartialEq, Eq)]
1460struct Url<'a> {
1461    /// `host[:port]` as written, for the Host header.
1462    authority: &'a str,
1463    /// The host for name resolution (brackets removed from IPv6 literals).
1464    host: &'a str,
1465    port: u16,
1466    /// Path and query; `/` when the URL has no path.
1467    target: &'a str,
1468}
1469
1470impl<'a> Url<'a> {
1471    fn parse(url: &'a str) -> Result<Self, Error> {
1472        let scheme_end = url.find("://").ok_or(Error::InvalidUrl("missing scheme"))?;
1473        if !url[..scheme_end].eq_ignore_ascii_case("http") {
1474            return Err(Error::InvalidUrl(
1475                "only http:// is supported (no TLS in this crate)",
1476            ));
1477        }
1478        let rest = &url[scheme_end + 3..];
1479        // The fragment is never sent.
1480        let rest = rest.split_once('#').map_or(rest, |(r, _)| r);
1481        let auth_end = rest.find(['/', '?']).unwrap_or(rest.len());
1482        let authority = &rest[..auth_end];
1483        let target = match &rest[auth_end..] {
1484            "" => "/",
1485            t if t.starts_with('?') => {
1486                return Err(Error::InvalidUrl("a query needs a path, for example /?q"));
1487            }
1488            t => t,
1489        };
1490        if authority.is_empty() {
1491            return Err(Error::InvalidUrl("missing host"));
1492        }
1493        if authority.contains('@') {
1494            return Err(Error::InvalidUrl(
1495                "credentials in the URL are not supported",
1496            ));
1497        }
1498        let (host, port) = if let Some(v6) = authority.strip_prefix('[') {
1499            let (inside, after) = v6.split_once(']').ok_or(Error::InvalidUrl("unclosed ["))?;
1500            let port = match after {
1501                "" => None,
1502                p => Some(
1503                    p.strip_prefix(':')
1504                        .ok_or(Error::InvalidUrl("junk after ]"))?,
1505                ),
1506            };
1507            (inside, port)
1508        } else {
1509            match authority.rsplit_once(':') {
1510                Some((h, p)) => (h, Some(p)),
1511                None => (authority, None),
1512            }
1513        };
1514        let port = match port {
1515            None => 80,
1516            Some(p) => p.parse().map_err(|_| Error::InvalidUrl("bad port"))?,
1517        };
1518        if host.is_empty() {
1519            return Err(Error::InvalidUrl("missing host"));
1520        }
1521        Ok(Self {
1522            authority,
1523            host,
1524            port,
1525            target,
1526        })
1527    }
1528}
1529
1530/// Formats `time` as an IMF-fixdate (RFC 9110 section 5.6.7), for example
1531/// `Sun, 06 Nov 1994 08:49:37 GMT`.
1532fn http_date(time: SystemTime) -> [u8; 29] {
1533    let secs = time.duration_since(UNIX_EPOCH).map_or(0, |d| d.as_secs());
1534    let days = secs / 86_400;
1535    let rem = secs % 86_400;
1536    let (year, month, day) = civil_from_days(days);
1537    const WEEKDAYS: [&[u8; 3]; 7] = [b"Thu", b"Fri", b"Sat", b"Sun", b"Mon", b"Tue", b"Wed"];
1538    const MONTHS: [&[u8; 3]; 12] = [
1539        b"Jan", b"Feb", b"Mar", b"Apr", b"May", b"Jun", b"Jul", b"Aug", b"Sep", b"Oct", b"Nov",
1540        b"Dec",
1541    ];
1542    let mut out = *b"Thu, 01 Jan 1970 00:00:00 GMT";
1543    out[..3].copy_from_slice(WEEKDAYS[(days % 7) as usize]);
1544    two_digits(&mut out[5..7], day);
1545    out[8..11].copy_from_slice(MONTHS[(month - 1) as usize]);
1546    let y = year.min(9999);
1547    two_digits(&mut out[12..14], y / 100);
1548    two_digits(&mut out[14..16], y % 100);
1549    two_digits(&mut out[17..19], rem / 3600);
1550    two_digits(&mut out[20..22], rem % 3600 / 60);
1551    two_digits(&mut out[23..25], rem % 60);
1552    out
1553}
1554
1555fn two_digits(dst: &mut [u8], v: u64) {
1556    if let [tens, ones] = dst {
1557        *tens = b'0' + (v / 10 % 10) as u8;
1558        *ones = b'0' + (v % 10) as u8;
1559    }
1560}
1561
1562/// Days since 1970-01-01 to (year, month 1..=12, day 1..=31), after Howard
1563/// Hinnant's `civil_from_days`, restricted to dates on or after the epoch.
1564fn civil_from_days(days: u64) -> (u64, u64, u64) {
1565    let z = days + 719_468;
1566    let era = z / 146_097;
1567    let doe = z - era * 146_097;
1568    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
1569    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
1570    let mp = (5 * doy + 2) / 153;
1571    let day = doy - (153 * mp + 2) / 5 + 1;
1572    let month = if mp < 10 { mp + 3 } else { mp - 9 };
1573    let year = yoe + era * 400 + u64::from(month <= 2);
1574    (year, month, day)
1575}
1576
1577#[cfg(test)]
1578mod tests {
1579    use super::*;
1580
1581    #[test]
1582    fn test_http_date_known_values() {
1583        let at = |s: u64| UNIX_EPOCH + Duration::from_secs(s);
1584        assert_eq!(&http_date(at(0)), b"Thu, 01 Jan 1970 00:00:00 GMT");
1585        assert_eq!(
1586            &http_date(at(784_111_777)),
1587            b"Sun, 06 Nov 1994 08:49:37 GMT"
1588        );
1589        assert_eq!(
1590            &http_date(at(951_782_400)),
1591            b"Tue, 29 Feb 2000 00:00:00 GMT"
1592        );
1593        assert_eq!(
1594            &http_date(at(1_791_072_000)),
1595            b"Sun, 04 Oct 2026 00:00:00 GMT"
1596        );
1597    }
1598
1599    #[test]
1600    fn test_url_parsing() {
1601        let u = Url::parse("http://127.0.0.1:8080/a/b?c=d#frag").unwrap();
1602        assert_eq!(
1603            u,
1604            Url {
1605                authority: "127.0.0.1:8080",
1606                host: "127.0.0.1",
1607                port: 8080,
1608                target: "/a/b?c=d"
1609            }
1610        );
1611        let u = Url::parse("HTTP://[::1]/").unwrap();
1612        assert_eq!((u.host, u.port, u.authority), ("::1", 80, "[::1]"));
1613        assert_eq!(Url::parse("http://example.com").unwrap().target, "/");
1614        for bad in [
1615            "example.com",
1616            "https://x/",
1617            "http:///x",
1618            "http://u:p@x/",
1619            "http://x:99999/",
1620            "http://[::1/",
1621            "http://[::1]x/",
1622            "http://x?q",
1623            "http://:80/",
1624        ] {
1625            assert!(
1626                matches!(Url::parse(bad), Err(Error::InvalidUrl(_))),
1627                "{bad}"
1628            );
1629        }
1630    }
1631
1632    #[test]
1633    fn test_error_display_and_source() {
1634        use std::error::Error as _;
1635        let e = Error::from(HttpError::BareCr);
1636        assert!(e.to_string().starts_with("HTTP error"));
1637        assert!(e.source().is_some());
1638        let e = Error::from(io::Error::other("x"));
1639        assert!(e.source().is_some());
1640        assert!(Error::InvalidUrl("y").source().is_none());
1641    }
1642}