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,
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 let status = e.status().unwrap_or(400);
677 let message = e.to_string();
678 if conn
679 .respond(&Response::text(status, &message).close())
680 .is_err()
681 {
682 return;
683 }
684 break;
685 }
686 Err(Error::Io(e)) if is_timeout(&e) && conn.request_in_progress() => {
687 // A request that did not arrive in time: say so, then
688 // close (the rest of it can no longer be framed).
689 if conn.send_request_timeout().is_ok() {
690 conn.close();
691 }
692 return;
693 }
694 Err(_) => return,
695 };
696 // `Server` writes full bodies only: the handler has no way to send
697 // a streamed one, so it would leave the client waiting. A `HEAD`
698 // answer is the exception: it announces a length or chunked
699 // framing but carries no body, so it is complete once written.
700 let streamed = response.body_bytes().is_none();
701 let response = if streamed && !is_head {
702 Response::text(
703 500,
704 "handler error: Server cannot send a streamed response; \
705 use Connection for streaming",
706 )
707 .close()
708 } else {
709 response
710 };
711 match conn.respond(&response) {
712 Ok(()) if streamed && is_head => {
713 if conn.end_body().is_err() {
714 return;
715 }
716 }
717 Ok(()) => {}
718 // Refused before anything was written (a builder error, or a
719 // 101 to a request that did not ask): answer 500 instead.
720 Err(Error::Http(e)) => {
721 let message = format!("handler error: invalid response: {e}");
722 if conn
723 .respond(&Response::text(500, &message).close())
724 .is_err()
725 {
726 return;
727 }
728 }
729 Err(_) => return,
730 }
731 if conn.is_upgraded() {
732 match (conn.into_upgraded(), upgrade) {
733 (Some(up), Some(hook)) => hook(up),
734 (Some(up), None) => drain_and_close(up.stream),
735 (None, _) => {}
736 }
737 return;
738 }
739 }
740 conn.close();
741}
742
743fn drain_and_close(mut stream: TcpStream) {
744 // Lingering close (RFC 9112 section 9.6): stop sending, then read what
745 // the peer is still sending so the close does not become a reset that
746 // discards the response before the peer reads it.
747 // The drain is bounded in bytes and in total time, not only per read:
748 // a peer that keeps trickling data must not hold the thread.
749 if stream.flush().is_err() || stream.shutdown(Shutdown::Write).is_err() {
750 return;
751 }
752 let deadline = Instant::now() + DRAIN_TOTAL;
753 let mut scratch = [0u8; 4096];
754 let mut drained = 0;
755 while drained < DRAIN_LIMIT {
756 let left = deadline.saturating_duration_since(Instant::now());
757 if left.is_zero()
758 || stream
759 .set_read_timeout(Some(left.min(DRAIN_READ_TIMEOUT)))
760 .is_err()
761 {
762 break;
763 }
764 match stream.read(&mut scratch) {
765 Ok(0) | Err(_) => break,
766 Ok(n) => drained += n,
767 }
768 }
769}
770
771fn is_timeout(e: &io::Error) -> bool {
772 // Unix reports an expired socket timeout as WouldBlock, Windows as
773 // TimedOut.
774 matches!(
775 e.kind(),
776 io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
777 )
778}
779
780/// One request read by [`Connection::next_request`]: its head and its
781/// complete body.
782///
783/// # Examples
784///
785/// See [`Connection`].
786#[derive(Debug)]
787pub struct Incoming<'a> {
788 head: RequestHead<'a>,
789 body: &'a [u8],
790}
791
792impl<'a> Incoming<'a> {
793 /// The request head.
794 ///
795 /// # Examples
796 ///
797 /// See [`Connection`].
798 #[must_use]
799 pub fn head(&self) -> RequestHead<'a> {
800 self.head
801 }
802
803 /// The complete request body (empty when there is none).
804 ///
805 /// # Examples
806 ///
807 /// See [`Connection`].
808 #[must_use]
809 pub fn body(&self) -> &'a [u8] {
810 self.body
811 }
812}
813
814/// A blocking server connection: [`ServerConn`] driven over a `TcpStream`.
815///
816/// Use it for your own accept loop, or to keep the socket after an
817/// upgrade with [`into_upgraded`](Self::into_upgraded).
818///
819/// # Examples
820///
821/// ```
822/// use std::net::{TcpListener, TcpStream};
823/// use std::io::{Read, Write};
824/// use http_net::{blocking::Connection, Limits, Response};
825///
826/// let listener = TcpListener::bind("127.0.0.1:0")?;
827/// let addr = listener.local_addr()?;
828/// let client = std::thread::spawn(move || -> std::io::Result<Vec<u8>> {
829/// let mut s = TcpStream::connect(addr)?;
830/// s.write_all(b"POST /sum HTTP/1.1\r\nHost: x\r\nContent-Length: 3\r\nConnection: close\r\n\r\n1 2")?;
831/// let mut reply = Vec::new();
832/// s.read_to_end(&mut reply)?;
833/// Ok(reply)
834/// });
835///
836/// let (stream, _) = listener.accept()?;
837/// let mut conn = Connection::new(stream, Limits::new());
838/// while let Some(req) = conn.next_request()? {
839/// let sum: u32 = std::str::from_utf8(req.body())?.split(' ').filter_map(|n| n.parse::<u32>().ok()).sum();
840/// let response = Response::text(200, &sum.to_string());
841/// conn.respond(&response)?;
842/// }
843/// conn.close();
844/// let reply = client.join().unwrap()?;
845/// assert!(reply.ends_with(b"\r\n\r\n3"));
846/// # Ok::<(), Box<dyn std::error::Error>>(())
847/// ```
848#[derive(Debug)]
849pub struct Connection {
850 stream: TcpStream,
851 conn: ServerConn,
852 body: Vec<u8>,
853 out: Vec<u8>,
854 /// The socket's own read timeout, restored once a request deadline no
855 /// longer applies.
856 read_timeout: Option<Duration>,
857 request_timeout: Option<Duration>,
858 /// When the request being read must be complete.
859 deadline: Option<Instant>,
860 /// Whether the socket's read timeout is currently shortened to the
861 /// deadline.
862 shortened: bool,
863}
864
865impl Connection {
866 /// Wraps an accepted socket. Set its timeouts first if you want them:
867 /// the read timeout set on `stream` is the idle timeout, and
868 /// [`with_request_timeout`](Self::with_request_timeout) adds a deadline
869 /// for each request as a whole.
870 ///
871 /// # Examples
872 ///
873 /// See [`Connection`].
874 #[must_use]
875 pub fn new(stream: TcpStream, limits: Limits) -> Self {
876 let read_timeout = stream.read_timeout().ok().flatten();
877 Self {
878 stream,
879 conn: ServerConn::with_limits(limits),
880 body: Vec::new(),
881 out: Vec::new(),
882 read_timeout,
883 request_timeout: None,
884 deadline: None,
885 shortened: false,
886 }
887 }
888
889 /// Sets a deadline for each request: from the moment its first byte is
890 /// seen to the end of its body (see
891 /// [`ServerConfig::with_request_timeout`]). When it passes,
892 /// [`next_request`](Self::next_request) fails with an [`Error::Io`] of
893 /// kind `TimedOut`. `None` (the default for a bare `Connection`) leaves
894 /// only the socket's own read timeout.
895 ///
896 /// # Examples
897 ///
898 /// ```no_run
899 /// use std::time::Duration;
900 /// # fn demo(stream: std::net::TcpStream) {
901 /// let conn = http_net::blocking::Connection::new(stream, http_net::Limits::new())
902 /// .with_request_timeout(Some(Duration::from_secs(10)));
903 /// # let _ = conn;
904 /// # }
905 /// ```
906 #[must_use]
907 pub fn with_request_timeout(mut self, timeout: Option<Duration>) -> Self {
908 self.request_timeout = timeout;
909 self
910 }
911
912 /// Whether part of a request has arrived and the rest is outstanding.
913 fn request_in_progress(&self) -> bool {
914 self.conn.request().is_some() || !self.conn.remaining_input().is_empty()
915 }
916
917 /// Answers a request that did not arrive in time with `408`, unless a
918 /// response is already on its way.
919 fn send_request_timeout(&mut self) -> Result<(), Error> {
920 if self.conn.response_started() {
921 return Err(Error::Http(HttpError::InvalidState(
922 "a response was already started",
923 )));
924 }
925 const BODY: &[u8] = b"request timed out";
926 self.out.clear();
927 self.out.extend_from_slice(
928 b"HTTP/1.1 408 Request Timeout\r\ncontent-type: text/plain; charset=utf-8\r\n\
929 content-length: ",
930 );
931 self.out
932 .extend_from_slice(BODY.len().to_string().as_bytes());
933 self.out
934 .extend_from_slice(b"\r\nconnection: close\r\ndate: ");
935 self.out.extend_from_slice(&http_date(SystemTime::now()));
936 self.out.extend_from_slice(b"\r\n\r\n");
937 self.out.extend_from_slice(BODY);
938 self.flush_out()
939 }
940
941 /// Reads into the parser's buffer, under the request deadline if one
942 /// is running.
943 fn fill(&mut self) -> Result<(), Error> {
944 if let Some(limit) = self.request_timeout {
945 if self.deadline.is_none() && self.request_in_progress() {
946 self.deadline = Some(Instant::now() + limit);
947 }
948 }
949 match self.deadline {
950 Some(deadline) => {
951 let left = deadline.saturating_duration_since(Instant::now());
952 if left.is_zero() {
953 return Err(Error::Io(io::Error::new(
954 io::ErrorKind::TimedOut,
955 "the request did not arrive within the request timeout",
956 )));
957 }
958 let wait = self.read_timeout.map_or(left, |t| t.min(left));
959 self.stream.set_read_timeout(Some(wait))?;
960 self.shortened = true;
961 }
962 None if self.shortened => {
963 self.stream.set_read_timeout(self.read_timeout)?;
964 self.shortened = false;
965 }
966 None => {}
967 }
968 let spare = self.conn.read_buf();
969 if spare.is_empty() {
970 return Err(Error::Http(HttpError::InvalidState(
971 "input buffer full while the parser needs input",
972 )));
973 }
974 let n = loop {
975 match self.stream.read(spare) {
976 Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
977 other => break other?,
978 }
979 };
980 if n == 0 {
981 self.conn.feed_eof();
982 } else {
983 self.conn.commit(n);
984 }
985 Ok(())
986 }
987
988 /// Reads the next complete request. `Ok(None)` means the connection is
989 /// finished (the client closed between requests, the previous exchange
990 /// ended with `connection: close`, or the connection was upgraded).
991 ///
992 /// A request that carries `Expect: 100-continue` is answered with
993 /// `100 Continue` before its body is read.
994 ///
995 /// # Errors
996 ///
997 /// [`Error::Http`] for a request this crate refuses (answer it with its
998 /// [`HttpError::status`] and close), [`Error::Io`] for socket failures
999 /// and timeouts (including the request deadline, as `TimedOut`), and
1000 /// [`Error::Http`] with [`HttpError::InvalidState`] if the previous
1001 /// request has not been answered.
1002 ///
1003 /// # Examples
1004 ///
1005 /// See [`Connection`].
1006 pub fn next_request(&mut self) -> Result<Option<Incoming<'_>>, Error> {
1007 self.body.clear();
1008 self.deadline = None;
1009 loop {
1010 match self.conn.poll()? {
1011 ServerEvent::Request(head) => {
1012 // The whole body is read before the caller sees the
1013 // request, so the expectation is answered here (RFC 9110
1014 // section 10.1.1). A body over the limit was already
1015 // refused from its Content-Length.
1016 if head.expects_continue() {
1017 self.out.clear();
1018 self.conn.send_continue(&mut self.out)?;
1019 self.flush_out()?;
1020 }
1021 }
1022 ServerEvent::Body(chunk) => self.body.extend_from_slice(chunk),
1023 ServerEvent::RequestEnd => break,
1024 ServerEvent::NeedInput => self.fill()?,
1025 ServerEvent::AwaitingResponse => {
1026 return Err(Error::Http(HttpError::InvalidState(
1027 "respond to the previous request first",
1028 )));
1029 }
1030 _ => return Ok(None),
1031 }
1032 }
1033 match self.conn.request() {
1034 Some(head) => Ok(Some(Incoming {
1035 head,
1036 body: &self.body,
1037 })),
1038 None => Ok(None),
1039 }
1040 }
1041
1042 /// Writes `response` to the socket. A streamed response continues with
1043 /// [`send_body`](Self::send_body) and [`end_body`](Self::end_body).
1044 ///
1045 /// # Errors
1046 ///
1047 /// [`Error::Http`] when the response is invalid or out of order (see
1048 /// [`ServerConn::respond`]); [`Error::Io`] when the write fails.
1049 ///
1050 /// # Examples
1051 ///
1052 /// See [`Connection`].
1053 pub fn respond(&mut self, response: &Response) -> Result<(), Error> {
1054 self.out.clear();
1055 let date = http_date(SystemTime::now());
1056 self.conn
1057 .respond_dated(response, Some(&date), &mut self.out)?;
1058 self.flush_out()
1059 }
1060
1061 /// Writes body bytes of a streamed response.
1062 ///
1063 /// # Errors
1064 ///
1065 /// As [`ServerConn::send_body`], plus [`Error::Io`].
1066 ///
1067 /// # Examples
1068 ///
1069 /// ```no_run
1070 /// # fn demo(conn: &mut http_net::blocking::Connection) -> Result<(), http_net::blocking::Error> {
1071 /// conn.respond(&http_net::Response::new(200).chunked())?;
1072 /// conn.send_body(b"first")?;
1073 /// conn.end_body()?;
1074 /// # Ok(()) }
1075 /// ```
1076 pub fn send_body(&mut self, data: &[u8]) -> Result<(), Error> {
1077 self.out.clear();
1078 self.conn.send_body(data, &mut self.out)?;
1079 self.flush_out()
1080 }
1081
1082 /// Ends a streamed response.
1083 ///
1084 /// # Errors
1085 ///
1086 /// As [`ServerConn::end_body`], plus [`Error::Io`].
1087 ///
1088 /// # Examples
1089 ///
1090 /// See [`send_body`](Self::send_body).
1091 pub fn end_body(&mut self) -> Result<(), Error> {
1092 self.out.clear();
1093 self.conn.end_body(&mut self.out)?;
1094 self.flush_out()
1095 }
1096
1097 fn flush_out(&mut self) -> Result<(), Error> {
1098 self.stream.write_all(&self.out)?;
1099 if self.out.capacity() > OUT_RETAIN {
1100 self.out = Vec::new();
1101 }
1102 Ok(())
1103 }
1104
1105 /// Whether a `101` (or a 2xx to `CONNECT`) was sent.
1106 ///
1107 /// # Examples
1108 ///
1109 /// See [`into_upgraded`](Self::into_upgraded).
1110 #[must_use]
1111 pub fn is_upgraded(&self) -> bool {
1112 self.conn.is_upgraded()
1113 }
1114
1115 /// The socket and the bytes of the new protocol, after an upgrade;
1116 /// `None` if the connection was not upgraded. The socket's read timeout
1117 /// is the one it had when the `Connection` was created (a request
1118 /// deadline may have shortened it while the request was read).
1119 ///
1120 /// # Examples
1121 ///
1122 /// ```no_run
1123 /// # fn demo(mut conn: http_net::blocking::Connection) {
1124 /// if conn.is_upgraded() {
1125 /// let up = conn.into_upgraded().unwrap();
1126 /// // speak the new protocol on up.stream, starting with up.leftover
1127 /// # let _ = up;
1128 /// }
1129 /// # }
1130 /// ```
1131 #[must_use]
1132 pub fn into_upgraded(self) -> Option<Upgraded> {
1133 if !self.is_upgraded() {
1134 return None;
1135 }
1136 let target = self
1137 .conn
1138 .request()
1139 .map(|r| r.target().to_string())
1140 .unwrap_or_default();
1141 if self.shortened {
1142 // Best effort: the new protocol must not inherit what was left
1143 // of the request deadline. Setting a timeout this socket has
1144 // already accepted does not fail in practice, and the hand-off
1145 // is more useful than refusing it over one.
1146 let _ = self.stream.set_read_timeout(self.read_timeout);
1147 }
1148 Some(Upgraded {
1149 leftover: self.conn.remaining_input().to_vec(),
1150 target,
1151 stream: self.stream,
1152 })
1153 }
1154
1155 /// Closes the connection gracefully: stops sending, briefly drains what
1156 /// the client is still sending, and drops the socket.
1157 ///
1158 /// # Examples
1159 ///
1160 /// See [`Connection`].
1161 pub fn close(self) {
1162 drain_and_close(self.stream);
1163 }
1164}
1165
1166/// A response fetched by [`get`] or [`Client::get`], fully read.
1167///
1168/// # Examples
1169///
1170/// See [`Server`].
1171#[derive(Clone, Debug, PartialEq, Eq)]
1172pub struct ClientResponse {
1173 status: u16,
1174 headers: Vec<(String, Vec<u8>)>,
1175 body: Vec<u8>,
1176}
1177
1178impl ClientResponse {
1179 /// The status code.
1180 ///
1181 /// # Examples
1182 ///
1183 /// See [`Server`].
1184 #[must_use]
1185 pub fn status(&self) -> u16 {
1186 self.status
1187 }
1188
1189 /// The fields in arrival order.
1190 ///
1191 /// # Examples
1192 ///
1193 /// ```no_run
1194 /// let resp = http_net::blocking::get("http://127.0.0.1:8080/")?;
1195 /// for (name, value) in resp.headers() {
1196 /// let _ = (name, value);
1197 /// }
1198 /// # Ok::<(), http_net::blocking::Error>(())
1199 /// ```
1200 pub fn headers(&self) -> impl Iterator<Item = (&str, &[u8])> {
1201 self.headers.iter().map(|(n, v)| (n.as_str(), v.as_slice()))
1202 }
1203
1204 /// The first field named `name` (ASCII case-insensitive).
1205 ///
1206 /// # Examples
1207 ///
1208 /// ```no_run
1209 /// let resp = http_net::blocking::get("http://127.0.0.1:8080/")?;
1210 /// let _ = resp.header("content-type");
1211 /// # Ok::<(), http_net::blocking::Error>(())
1212 /// ```
1213 #[must_use]
1214 pub fn header(&self, name: &str) -> Option<&[u8]> {
1215 self.headers
1216 .iter()
1217 .find(|(n, _)| n.eq_ignore_ascii_case(name))
1218 .map(|(_, v)| v.as_slice())
1219 }
1220
1221 /// The body.
1222 ///
1223 /// # Examples
1224 ///
1225 /// See [`Server`].
1226 #[must_use]
1227 pub fn body(&self) -> &[u8] {
1228 &self.body
1229 }
1230
1231 /// The body, if it is UTF-8.
1232 ///
1233 /// # Examples
1234 ///
1235 /// See [`Server`].
1236 #[must_use]
1237 pub fn text(&self) -> Option<&str> {
1238 core::str::from_utf8(&self.body).ok()
1239 }
1240
1241 /// Takes the body.
1242 ///
1243 /// # Examples
1244 ///
1245 /// ```no_run
1246 /// let bytes = http_net::blocking::get("http://127.0.0.1:8080/")?.into_body();
1247 /// # let _ = bytes;
1248 /// # Ok::<(), http_net::blocking::Error>(())
1249 /// ```
1250 #[must_use]
1251 pub fn into_body(self) -> Vec<u8> {
1252 self.body
1253 }
1254}
1255
1256/// Fetches `url` with `GET` using the default [`Client`].
1257///
1258/// # Errors
1259///
1260/// [`Error::InvalidUrl`] for anything but `http://host[:port][/path]`,
1261/// [`Error::Io`] for connection failures and timeouts, [`Error::Http`] for
1262/// a response this crate refuses (including one over the body limit).
1263///
1264/// # Examples
1265///
1266/// See [`Server`].
1267pub fn get(url: &str) -> Result<ClientResponse, Error> {
1268 Client::new().get(url)
1269}
1270
1271/// A blocking HTTP/1.1 client: one connection per request.
1272///
1273/// | Setting | Default |
1274/// |---|---|
1275/// | limits | [`Limits::new`] with a 16 MiB body |
1276/// | connect, read, and write timeout | 30 s each |
1277///
1278/// # Examples
1279///
1280/// ```no_run
1281/// use std::time::Duration;
1282/// use http_net::{blocking::Client, Limits};
1283///
1284/// let client = Client::new()
1285/// .with_limits(Limits::new().with_max_body(64 << 20))
1286/// .with_timeout(Duration::from_secs(5));
1287/// let resp = client.get("http://127.0.0.1:8080/big")?;
1288/// # let _ = resp;
1289/// # Ok::<(), http_net::blocking::Error>(())
1290/// ```
1291#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1292pub struct Client {
1293 limits: Limits,
1294 timeout: Duration,
1295}
1296
1297impl Default for Client {
1298 fn default() -> Self {
1299 Self::new()
1300 }
1301}
1302
1303impl Client {
1304 /// The defaults (see the table on [`Client`]).
1305 ///
1306 /// # Examples
1307 ///
1308 /// ```
1309 /// let _ = http_net::blocking::Client::new();
1310 /// ```
1311 #[must_use]
1312 pub const fn new() -> Self {
1313 Self {
1314 limits: Limits::new().with_max_body(16 * 1024 * 1024),
1315 timeout: Duration::from_secs(30),
1316 }
1317 }
1318
1319 /// Sets the response limits.
1320 ///
1321 /// # Examples
1322 ///
1323 /// ```
1324 /// let _ = http_net::blocking::Client::new().with_limits(http_net::Limits::new());
1325 /// ```
1326 #[must_use]
1327 pub const fn with_limits(mut self, limits: Limits) -> Self {
1328 self.limits = limits;
1329 self
1330 }
1331
1332 /// Sets the connect, read, and write timeout (zero is raised to 1 ms).
1333 ///
1334 /// # Examples
1335 ///
1336 /// ```
1337 /// let _ = http_net::blocking::Client::new().with_timeout(std::time::Duration::from_secs(2));
1338 /// ```
1339 #[must_use]
1340 pub const fn with_timeout(mut self, timeout: Duration) -> Self {
1341 self.timeout = if timeout.is_zero() {
1342 Duration::from_millis(1)
1343 } else {
1344 timeout
1345 };
1346 self
1347 }
1348
1349 /// Fetches `url` with `GET`.
1350 ///
1351 /// # Errors
1352 ///
1353 /// As [`get`].
1354 ///
1355 /// # Examples
1356 ///
1357 /// See [`Client`].
1358 pub fn get(&self, url: &str) -> Result<ClientResponse, Error> {
1359 let url = Url::parse(url)?;
1360 let request = Request::get(url.target)
1361 .header("host", url.authority)
1362 .close();
1363 let mut stream = self.connect(&url)?;
1364 let mut conn = ClientConn::with_limits(self.limits);
1365 let mut out = Vec::new();
1366 conn.send_request(&request, &mut out)?;
1367 stream.write_all(&out)?;
1368 read_response(&mut stream, &mut conn)
1369 }
1370
1371 fn connect(&self, url: &Url<'_>) -> Result<TcpStream, Error> {
1372 let mut last = None;
1373 for addr in (url.host, url.port).to_socket_addrs()? {
1374 match TcpStream::connect_timeout(&addr, self.timeout) {
1375 Ok(stream) => {
1376 stream.set_read_timeout(Some(self.timeout))?;
1377 stream.set_write_timeout(Some(self.timeout))?;
1378 stream.set_nodelay(true)?;
1379 return Ok(stream);
1380 }
1381 Err(e) => last = Some(e),
1382 }
1383 }
1384 Err(Error::Io(last.unwrap_or_else(|| {
1385 io::Error::new(io::ErrorKind::NotFound, "the host resolved to no address")
1386 })))
1387 }
1388}
1389
1390fn read_response(stream: &mut TcpStream, conn: &mut ClientConn) -> Result<ClientResponse, Error> {
1391 let mut response = ClientResponse {
1392 status: 0,
1393 headers: Vec::new(),
1394 body: Vec::new(),
1395 };
1396 loop {
1397 match conn.poll()? {
1398 ClientEvent::Response(head) => {
1399 response.status = head.status();
1400 response.headers = head
1401 .headers()
1402 .map(|(n, v)| (n.to_string(), v.to_vec()))
1403 .collect();
1404 }
1405 ClientEvent::Body(chunk) => response.body.extend_from_slice(chunk),
1406 ClientEvent::ResponseEnd => return Ok(response),
1407 ClientEvent::NeedInput => {
1408 let spare = conn.read_buf();
1409 if spare.is_empty() {
1410 return Err(Error::Http(HttpError::InvalidState(
1411 "input buffer full while the parser needs input",
1412 )));
1413 }
1414 let n = loop {
1415 match stream.read(spare) {
1416 Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
1417 other => break other?,
1418 }
1419 };
1420 if n == 0 {
1421 conn.feed_eof();
1422 } else {
1423 conn.commit(n);
1424 }
1425 }
1426 _ => {
1427 return Err(Error::Http(HttpError::InvalidState(
1428 "the connection ended without a complete response",
1429 )));
1430 }
1431 }
1432 }
1433}
1434
1435/// The parts of an `http://` URL needed to send a request.
1436#[derive(Debug, PartialEq, Eq)]
1437struct Url<'a> {
1438 /// `host[:port]` as written, for the Host header.
1439 authority: &'a str,
1440 /// The host for name resolution (brackets removed from IPv6 literals).
1441 host: &'a str,
1442 port: u16,
1443 /// Path and query; `/` when the URL has no path.
1444 target: &'a str,
1445}
1446
1447impl<'a> Url<'a> {
1448 fn parse(url: &'a str) -> Result<Self, Error> {
1449 let scheme_end = url.find("://").ok_or(Error::InvalidUrl("missing scheme"))?;
1450 if !url[..scheme_end].eq_ignore_ascii_case("http") {
1451 return Err(Error::InvalidUrl(
1452 "only http:// is supported (no TLS in this crate)",
1453 ));
1454 }
1455 let rest = &url[scheme_end + 3..];
1456 // The fragment is never sent.
1457 let rest = rest.split_once('#').map_or(rest, |(r, _)| r);
1458 let auth_end = rest.find(['/', '?']).unwrap_or(rest.len());
1459 let authority = &rest[..auth_end];
1460 let target = match &rest[auth_end..] {
1461 "" => "/",
1462 t if t.starts_with('?') => {
1463 return Err(Error::InvalidUrl("a query needs a path, for example /?q"));
1464 }
1465 t => t,
1466 };
1467 if authority.is_empty() {
1468 return Err(Error::InvalidUrl("missing host"));
1469 }
1470 if authority.contains('@') {
1471 return Err(Error::InvalidUrl(
1472 "credentials in the URL are not supported",
1473 ));
1474 }
1475 let (host, port) = if let Some(v6) = authority.strip_prefix('[') {
1476 let (inside, after) = v6.split_once(']').ok_or(Error::InvalidUrl("unclosed ["))?;
1477 let port = match after {
1478 "" => None,
1479 p => Some(
1480 p.strip_prefix(':')
1481 .ok_or(Error::InvalidUrl("junk after ]"))?,
1482 ),
1483 };
1484 (inside, port)
1485 } else {
1486 match authority.rsplit_once(':') {
1487 Some((h, p)) => (h, Some(p)),
1488 None => (authority, None),
1489 }
1490 };
1491 let port = match port {
1492 None => 80,
1493 Some(p) => p.parse().map_err(|_| Error::InvalidUrl("bad port"))?,
1494 };
1495 if host.is_empty() {
1496 return Err(Error::InvalidUrl("missing host"));
1497 }
1498 Ok(Self {
1499 authority,
1500 host,
1501 port,
1502 target,
1503 })
1504 }
1505}
1506
1507/// Formats `time` as an IMF-fixdate (RFC 9110 section 5.6.7), for example
1508/// `Sun, 06 Nov 1994 08:49:37 GMT`.
1509fn http_date(time: SystemTime) -> [u8; 29] {
1510 let secs = time.duration_since(UNIX_EPOCH).map_or(0, |d| d.as_secs());
1511 let days = secs / 86_400;
1512 let rem = secs % 86_400;
1513 let (year, month, day) = civil_from_days(days);
1514 const WEEKDAYS: [&[u8; 3]; 7] = [b"Thu", b"Fri", b"Sat", b"Sun", b"Mon", b"Tue", b"Wed"];
1515 const MONTHS: [&[u8; 3]; 12] = [
1516 b"Jan", b"Feb", b"Mar", b"Apr", b"May", b"Jun", b"Jul", b"Aug", b"Sep", b"Oct", b"Nov",
1517 b"Dec",
1518 ];
1519 let mut out = *b"Thu, 01 Jan 1970 00:00:00 GMT";
1520 out[..3].copy_from_slice(WEEKDAYS[(days % 7) as usize]);
1521 two_digits(&mut out[5..7], day);
1522 out[8..11].copy_from_slice(MONTHS[(month - 1) as usize]);
1523 let y = year.min(9999);
1524 two_digits(&mut out[12..14], y / 100);
1525 two_digits(&mut out[14..16], y % 100);
1526 two_digits(&mut out[17..19], rem / 3600);
1527 two_digits(&mut out[20..22], rem % 3600 / 60);
1528 two_digits(&mut out[23..25], rem % 60);
1529 out
1530}
1531
1532fn two_digits(dst: &mut [u8], v: u64) {
1533 if let [tens, ones] = dst {
1534 *tens = b'0' + (v / 10 % 10) as u8;
1535 *ones = b'0' + (v % 10) as u8;
1536 }
1537}
1538
1539/// Days since 1970-01-01 to (year, month 1..=12, day 1..=31), after Howard
1540/// Hinnant's `civil_from_days`, restricted to dates on or after the epoch.
1541fn civil_from_days(days: u64) -> (u64, u64, u64) {
1542 let z = days + 719_468;
1543 let era = z / 146_097;
1544 let doe = z - era * 146_097;
1545 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
1546 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
1547 let mp = (5 * doy + 2) / 153;
1548 let day = doy - (153 * mp + 2) / 5 + 1;
1549 let month = if mp < 10 { mp + 3 } else { mp - 9 };
1550 let year = yoe + era * 400 + u64::from(month <= 2);
1551 (year, month, day)
1552}
1553
1554#[cfg(test)]
1555mod tests {
1556 use super::*;
1557
1558 #[test]
1559 fn test_http_date_known_values() {
1560 let at = |s: u64| UNIX_EPOCH + Duration::from_secs(s);
1561 assert_eq!(&http_date(at(0)), b"Thu, 01 Jan 1970 00:00:00 GMT");
1562 assert_eq!(
1563 &http_date(at(784_111_777)),
1564 b"Sun, 06 Nov 1994 08:49:37 GMT"
1565 );
1566 assert_eq!(
1567 &http_date(at(951_782_400)),
1568 b"Tue, 29 Feb 2000 00:00:00 GMT"
1569 );
1570 assert_eq!(
1571 &http_date(at(1_791_072_000)),
1572 b"Sun, 04 Oct 2026 00:00:00 GMT"
1573 );
1574 }
1575
1576 #[test]
1577 fn test_url_parsing() {
1578 let u = Url::parse("http://127.0.0.1:8080/a/b?c=d#frag").unwrap();
1579 assert_eq!(
1580 u,
1581 Url {
1582 authority: "127.0.0.1:8080",
1583 host: "127.0.0.1",
1584 port: 8080,
1585 target: "/a/b?c=d"
1586 }
1587 );
1588 let u = Url::parse("HTTP://[::1]/").unwrap();
1589 assert_eq!((u.host, u.port, u.authority), ("::1", 80, "[::1]"));
1590 assert_eq!(Url::parse("http://example.com").unwrap().target, "/");
1591 for bad in [
1592 "example.com",
1593 "https://x/",
1594 "http:///x",
1595 "http://u:p@x/",
1596 "http://x:99999/",
1597 "http://[::1/",
1598 "http://[::1]x/",
1599 "http://x?q",
1600 "http://:80/",
1601 ] {
1602 assert!(
1603 matches!(Url::parse(bad), Err(Error::InvalidUrl(_))),
1604 "{bad}"
1605 );
1606 }
1607 }
1608
1609 #[test]
1610 fn test_error_display_and_source() {
1611 use std::error::Error as _;
1612 let e = Error::from(HttpError::BareCr);
1613 assert!(e.to_string().starts_with("HTTP error"));
1614 assert!(e.source().is_some());
1615 let e = Error::from(io::Error::other("x"));
1616 assert!(e.source().is_some());
1617 assert!(Error::InvalidUrl("y").source().is_none());
1618 }
1619}