From a83f7053287dde585fef87b3b5369f36d37034b8 Mon Sep 17 00:00:00 2001 From: Alex Gleason Date: Wed, 30 Sep 2026 15:52:27 -0500 Subject: [PATCH] feat(http1): send response bodies from files with `Write::poll_write_file` Adds `hyper::ext::SendFile`, a response extension naming a range of a file, and two provided methods on `rt::Write`: `supports_write_file` and `poll_write_file`. When a response carries a `SendFile`, the connection's IO supports it, and the head is framed by a `Content-Length` equal to the range, the HTTP/1 server writes the head and then hands the file to `poll_write_file` instead of polling the body. That lets an IO use a zero-copy system call such as `sendfile(2)`, so a static file's bytes go from the page cache to the socket without passing through the process. The body is still required and is the fallback: it is used unchanged for IO without support (TLS, and every existing IO, since the defaults opt out), for HTTP/2, and for responses whose framing doesn't match (chunked, HEAD, 304). When the file is sent, the body is dropped unpolled. Both trait methods have defaults, and the extension is new, so nothing existing changes behavior or stops compiling. Refs #3026 --- Cargo.toml | 5 + src/common/io/rewind.rs | 14 ++ src/ext/mod.rs | 6 + src/ext/send_file.rs | 80 ++++++ src/proto/h1/conn.rs | 81 ++++++ src/proto/h1/dispatch.rs | 29 ++- src/proto/h1/encode.rs | 22 ++ src/proto/h1/io.rs | 25 ++ src/rt/io.rs | 71 ++++++ tests/h1_send_file.rs | 515 +++++++++++++++++++++++++++++++++++++++ 10 files changed, 846 insertions(+), 2 deletions(-) create mode 100644 src/ext/send_file.rs create mode 100644 tests/h1_send_file.rs diff --git a/Cargo.toml b/Cargo.toml index edf0af719d..510ba71167 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -342,6 +342,11 @@ name = "server" path = "tests/server.rs" required-features = ["full"] +[[test]] +name = "h1_send_file" +path = "tests/h1_send_file.rs" +required-features = ["full"] + [[test]] name = "h1_shutdown_while_buffered" path = "tests/h1_shutdown_while_buffered.rs" diff --git a/src/common/io/rewind.rs b/src/common/io/rewind.rs index 7acd809685..b295fb27f6 100644 --- a/src/common/io/rewind.rs +++ b/src/common/io/rewind.rs @@ -107,6 +107,20 @@ where fn is_write_vectored(&self) -> bool { self.inner.is_write_vectored() } + + fn supports_write_file(&self) -> bool { + self.inner.supports_write_file() + } + + fn poll_write_file( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + file: &std::fs::File, + offset: u64, + len: usize, + ) -> Poll> { + Pin::new(&mut self.inner).poll_write_file(cx, file, offset, len) + } } #[cfg(all( diff --git a/src/ext/mod.rs b/src/ext/mod.rs index ba6886b835..c7a8db596f 100644 --- a/src/ext/mod.rs +++ b/src/ext/mod.rs @@ -27,6 +27,7 @@ //! The extensions in this module can be grouped as follows: //! //! - **HTTP/1 Reason Phrase**: [`ReasonPhrase`] — Access non-canonical reason phrases in HTTP/1 responses. +//! - **HTTP/1 File Bodies**: [`SendFile`] — Send a response body straight from a file, with zero-copy IO where the connection supports it. //! - **Informational Responses**: [`on_informational`] — Register callbacks for 1xx HTTP/1 responses on the client. //! - **Header Case Tracking**: Internal types for tracking the original casing and order of headers as received. //! - **HTTP/2 Protocol Extensions**: [`Protocol`] — Access the `:protocol` pseudo-header for Extended CONNECT in HTTP/2. @@ -54,6 +55,11 @@ mod h1_reason_phrase; #[cfg(any(feature = "http1", feature = "ffi"))] pub use h1_reason_phrase::ReasonPhrase; +#[cfg(all(feature = "http1", feature = "server"))] +mod send_file; +#[cfg(all(feature = "http1", feature = "server"))] +pub use send_file::SendFile; + #[cfg(all(feature = "http1", feature = "client"))] mod informational; #[cfg(all(feature = "http1", feature = "client"))] diff --git a/src/ext/send_file.rs b/src/ext/send_file.rs new file mode 100644 index 0000000000..d3e859d36f --- /dev/null +++ b/src/ext/send_file.rs @@ -0,0 +1,80 @@ +use std::fs::File; +use std::sync::Arc; + +/// A range of a file to send as an HTTP/1 response body without copying it +/// through the process. +/// +/// Insert a `SendFile` into the extensions of an `http::Response`, and hyper's +/// HTTP/1 server will write that range of the file to the connection with +/// [`Write::poll_write_file`](crate::rt::Write::poll_write_file) instead of +/// polling the response body. On a plain TCP socket this lets the IO use a +/// zero-copy system call like Linux's `sendfile(2)`, which hands the file's +/// pages from the kernel's page cache to the socket. +/// +/// The response body is still required, and must hold the same bytes. It is +/// the fallback, used whenever the file can't be sent directly: +/// +/// - the connection's IO doesn't support it (its +/// [`supports_write_file`](crate::rt::Write::supports_write_file) returns +/// `false`, as it does for TLS streams and by default), +/// - the connection is HTTP/2, +/// - or the response isn't framed with a `Content-Length` equal to `len`: +/// it is chunked, or has no body at all (a reply to `HEAD`, a `304`). +/// +/// When the file is sent, the body is dropped without being polled, so a body +/// that reads the file lazily costs nothing. +/// +/// The head is flushed on its own before the file is handed to the IO, so a +/// response sent this way costs at least one more write than a buffered one. +/// For small bodies that can outweigh the copy it saves; it pays off for +/// large ones. +/// +/// The file must not be modified while it is being sent. hyper reads it by +/// offset and never moves its cursor, so one `File` can be shared by any +/// number of responses at once. +/// +/// # Example +/// +/// ``` +/// # fn doc(file: std::sync::Arc, bytes: bytes::Bytes) { +/// use http_body_util::Full; +/// use hyper::ext::SendFile; +/// +/// let len = bytes.len() as u64; +/// let mut response = http::Response::new(Full::new(bytes)); +/// response.extensions_mut().insert(SendFile::new(file, 0, len)); +/// # } +/// ``` +#[derive(Clone, Debug)] +pub struct SendFile { + pub(crate) file: Arc, + pub(crate) offset: u64, + pub(crate) len: u64, +} + +impl SendFile { + /// Send `len` bytes of `file`, starting at `offset`. + pub fn new(file: Arc, offset: u64, len: u64) -> Self { + SendFile { file, offset, len } + } + + /// The file to send from. + pub fn file(&self) -> &Arc { + &self.file + } + + /// Where in the file the body starts. + pub fn offset(&self) -> u64 { + self.offset + } + + /// How many bytes of the file make up the body. + pub fn len(&self) -> u64 { + self.len + } + + /// Whether the body is empty. + pub fn is_empty(&self) -> bool { + self.len == 0 + } +} diff --git a/src/proto/h1/conn.rs b/src/proto/h1/conn.rs index 69ca50b1a4..0030a5d161 100644 --- a/src/proto/h1/conn.rs +++ b/src/proto/h1/conn.rs @@ -78,6 +78,8 @@ where notify_read: false, reading: Reading::Init, writing: Writing::Init, + #[cfg(feature = "server")] + send_file: None, upgrade: None, // We assume a modern world where the remote speaks HTTP/1.1. // If they tell us otherwise, we'll downgrade in `read_head`. @@ -822,6 +824,81 @@ where } } + /// Take a response's `SendFile`, if the IO can write it. + #[cfg(feature = "server")] + pub(crate) fn take_send_file( + &mut self, + head: &mut MessageHead, + ) -> Option { + let file = head.extensions.remove::()?; + if T::is_server() && self.io.supports_write_file() { + Some(file) + } else { + None + } + } + + /// Send the body just started by `write_head` from `file`, if its framing + /// allows: a `Content-Length` that is exactly the file's range. Anything + /// else (chunked, or no body at all) leaves the body to be polled. + #[cfg(feature = "server")] + pub(crate) fn start_send_file(&mut self, file: crate::ext::SendFile) -> bool { + match &self.state.writing { + Writing::Body(encoder) if encoder.remaining_length() == Some(file.len) => { + self.state.send_file = Some(file); + true + } + _ => false, + } + } + + #[cfg(feature = "server")] + pub(crate) fn is_sending_file(&self) -> bool { + self.state.send_file.is_some() && matches!(self.state.writing, Writing::Body(_)) + } + + /// Write the file started by `start_send_file` until it is done or the IO + /// would block. + #[cfg(feature = "server")] + pub(crate) fn poll_send_file(&mut self, cx: &mut Context<'_>) -> Poll> { + loop { + let (file, encoder) = if let (Some(file), Writing::Body(encoder)) = + (&mut self.state.send_file, &mut self.state.writing) + { + (file, encoder) + } else { + self.state.send_file = None; + return Poll::Ready(Ok(())); + }; + + let remaining = encoder.remaining_length().unwrap_or(0); + let len = usize::try_from(remaining).unwrap_or(usize::MAX); + let n = ready!(self + .io + .poll_write_file(cx, &file.file, file.offset, len))?; + if n == 0 { + // Either the peer stopped accepting bytes, or the file is + // shorter than the Content-Length already sent for it. + self.state.send_file = None; + self.state.writing = Writing::Closed; + return Poll::Ready(Err(io::ErrorKind::WriteZero.into())); + } + trace!("sent {} bytes from file", n); + file.offset += n as u64; + encoder.wrote_length(n as u64); + + if encoder.is_eof() { + self.state.writing = if encoder.is_last() { + Writing::Closed + } else { + Writing::KeepAlive + }; + self.state.send_file = None; + return Poll::Ready(Ok(())); + } + } + } + // When we get a parse error, depending on what side we are, we might be able // to write a response before closing the connection. // @@ -973,6 +1050,10 @@ struct State { reading: Reading, /// State of allowed writes. writing: Writing, + /// The file a `Writing::Body` is being sent from, when the IO writes + /// it directly rather than the body being polled. + #[cfg(feature = "server")] + send_file: Option, /// An expected pending HTTP upgrade. upgrade: Option, /// Either HTTP/1.0 or 1.1 connection. diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs index d98aebcf1f..2f91c0a182 100644 --- a/src/proto/h1/dispatch.rs +++ b/src/proto/h1/dispatch.rs @@ -362,7 +362,10 @@ where && self.dispatch.should_poll() { if let Some(msg) = ready!(Pin::new(&mut self.dispatch).poll_msg(cx)) { - let (head, body) = msg.map_err(crate::Error::new_user_service)?; + #[cfg_attr(not(feature = "server"), allow(unused_mut))] + let (mut head, body) = msg.map_err(crate::Error::new_user_service)?; + #[cfg(feature = "server")] + let send_file = self.conn.take_send_file(&mut head); let body_type = if body.is_end_stream() { self.body_rx.set(None); @@ -377,10 +380,22 @@ where btype }; self.conn.write_head(head, body_type); + + // The body stays the fallback until the head's framing is + // known to match the file; once it does, it's never read. + #[cfg(feature = "server")] + if let Some(file) = send_file { + if self.conn.start_send_file(file) { + self.body_rx.set(None); + } + } } else { self.close(); return Poll::Ready(Ok(())); } + } else if self.is_sending_file() { + #[cfg(feature = "server")] + ready!(self.conn.poll_send_file(cx)).map_err(crate::Error::new_body_write)?; } else if !self.conn.can_buffer_body() { ready!(self.poll_flush(cx))?; } else { @@ -466,6 +481,14 @@ where !self.is_closing && self.body_rx.is_some() && self.conn.can_write_body() } + /// Whether a body is being sent from a file, rather than from `body_rx`. + fn is_sending_file(&self) -> bool { + #[cfg(feature = "server")] + return self.conn.is_sending_file(); + #[cfg(not(feature = "server"))] + return false; + } + fn is_done(&self) -> bool { if self.is_closing { return true; @@ -478,7 +501,9 @@ where true } else { let write_done = self.conn.is_write_closed() - || (!self.dispatch.should_poll() && self.body_rx.is_none()); + || (!self.dispatch.should_poll() + && self.body_rx.is_none() + && !self.is_sending_file()); read_done && write_done } } diff --git a/src/proto/h1/encode.rs b/src/proto/h1/encode.rs index 266b4dd6dd..9a6123b65f 100644 --- a/src/proto/h1/encode.rs +++ b/src/proto/h1/encode.rs @@ -113,6 +113,28 @@ impl Encoder { matches!(self.kind, Kind::Chunked(_)) } + /// The bytes still owed to a body framed by `Content-Length`, or `None` + /// for any other framing. + #[cfg(feature = "server")] + pub(crate) fn remaining_length(&self) -> Option { + match self.kind { + Kind::Length(remaining) => Some(remaining), + _ => None, + } + } + + /// Count `n` body bytes the IO wrote on its own, without `encode`. + #[cfg(feature = "server")] + pub(crate) fn wrote_length(&mut self, n: u64) { + match &mut self.kind { + Kind::Length(remaining) => { + debug_assert!(n <= *remaining, "wrote past Content-Length"); + *remaining -= n; + } + _ => unreachable!("wrote_length on a body not framed by Content-Length"), + } + } + pub(crate) fn end(&self) -> Result>, NotEof> { match self.kind { Kind::Length(0) => Ok(None), diff --git a/src/proto/h1/io.rs b/src/proto/h1/io.rs index 5833004768..4f4fe42880 100644 --- a/src/proto/h1/io.rs +++ b/src/proto/h1/io.rs @@ -327,6 +327,31 @@ where Pin::new(&mut self.io).poll_flush(cx) } + #[cfg(feature = "server")] + pub(crate) fn supports_write_file(&self) -> bool { + self.io.supports_write_file() + } + + /// Write part of a file straight to the IO, after everything buffered. + #[cfg(feature = "server")] + pub(crate) fn poll_write_file( + &mut self, + cx: &mut Context<'_>, + file: &std::fs::File, + offset: u64, + len: usize, + ) -> Poll> { + // The head goes out first, since the file's bytes follow it on the + // wire. That holds even when pipelined flushes are being deferred. + if self.write_buf.remaining() > 0 { + let flush_pipeline = std::mem::replace(&mut self.flush_pipeline, false); + let flushed = self.poll_flush(cx); + self.flush_pipeline = flush_pipeline; + ready!(flushed)?; + } + Pin::new(&mut self.io).poll_write_file(cx, file, offset, len) + } + pub(crate) fn poll_shutdown(&mut self, cx: &mut Context<'_>) -> Poll> { ready!(self.poll_flush(cx))?; Pin::new(&mut self.io).poll_shutdown(cx) diff --git a/src/rt/io.rs b/src/rt/io.rs index 31e699d73a..c58190a948 100644 --- a/src/rt/io.rs +++ b/src/rt/io.rs @@ -143,6 +143,49 @@ pub trait Write { .map_or(&[][..], |b| &**b); self.poll_write(cx, buf) } + + /// Returns whether this writer has a `poll_write_file` implementation. + /// + /// hyper only calls `poll_write_file` on writers that return `true`. + /// + /// The default implementation returns `false`. + fn supports_write_file(&self) -> bool { + false + } + + /// Attempt to write up to `len` bytes of `file`, starting at `offset`, + /// without passing them through a buffer in this process. + /// + /// This is the hook for a zero-copy system call such as Linux's + /// `sendfile(2)`: on a plain socket, the kernel can move a file's bytes + /// from its page cache to the socket directly. hyper's HTTP/1 server uses + /// it for response bodies sent with `hyper::ext::SendFile`. + /// + /// On success, returns `Poll::Ready(Ok(num_bytes_written))`, where + /// `num_bytes_written <= len`. A return value of `0` means that the + /// file ended early or the destination can no longer accept bytes. + /// Implementations must not move the file's cursor, since a file may be + /// shared by many connections at once. + /// + /// If the object is not ready for writing, the method returns + /// `Poll::Pending` and arranges for the current task (via `cx.waker()`) to + /// receive a notification when the object becomes writable or is closed. + /// + /// The default implementation returns an error of kind + /// [`Unsupported`](std::io::ErrorKind::Unsupported). + fn poll_write_file( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + file: &std::fs::File, + offset: u64, + len: usize, + ) -> Poll> { + let _ = (cx, file, offset, len); + Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "poll_write_file is not supported by this writer", + ))) + } } /// A wrapper around a byte buffer that is incrementally filled and initialized. @@ -485,6 +528,20 @@ macro_rules! deref_async_write { (**self).is_write_vectored() } + fn supports_write_file(&self) -> bool { + (**self).supports_write_file() + } + + fn poll_write_file( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + file: &std::fs::File, + offset: u64, + len: usize, + ) -> Poll> { + Pin::new(&mut **self).poll_write_file(cx, file, offset, len) + } + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { Pin::new(&mut **self).poll_flush(cx) } @@ -531,6 +588,20 @@ where (**self).is_write_vectored() } + fn supports_write_file(&self) -> bool { + (**self).supports_write_file() + } + + fn poll_write_file( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + file: &std::fs::File, + offset: u64, + len: usize, + ) -> Poll> { + pin_as_deref_mut(self).poll_write_file(cx, file, offset, len) + } + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { pin_as_deref_mut(self).poll_flush(cx) } diff --git a/tests/h1_send_file.rs b/tests/h1_send_file.rs new file mode 100644 index 0000000000..bf57f63479 --- /dev/null +++ b/tests/h1_send_file.rs @@ -0,0 +1,515 @@ +#![cfg(unix)] +// Tests for `hyper::ext::SendFile`: an HTTP/1 response body written by the +// IO's `poll_write_file` rather than polled from the body. +// +// The IO here implements `poll_write_file` by reading a few bytes of the file +// at the offset and writing them to the socket, and it returns `Pending` before +// every such write, so a body always takes several calls across several polls +// and exercises the resumption bookkeeping. The fallback body is given +// different bytes from the file, so a response shows which one was sent. + +use std::convert::Infallible; +use std::fs::File; +use std::io; +use std::os::unix::fs::FileExt; +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::task::{Context, Poll}; +use std::time::Duration; + +use bytes::Bytes; +use futures_util::stream; +use http_body_util::{Full, StreamBody}; +use hyper::body::Frame; +use hyper::ext::SendFile; +use hyper::rt::{Read, ReadBufCursor, Write}; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::Response; +use support::TokioIo; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::time::timeout; + +mod support; + +const FILE: &[u8] = b"hello from the file"; +const FALLBACK: &[u8] = b"a fallback body...."; + +/// How many bytes one `poll_write_file` call sends at most. +const STEP: usize = 4; + +struct FileIo { + inner: TokioIo, + supports: bool, + file_writes: Arc, + ready: bool, +} + +impl Read for FileIo { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: ReadBufCursor<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl Write for FileIo { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } + + fn supports_write_file(&self) -> bool { + self.supports + } + + fn poll_write_file( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + file: &File, + offset: u64, + len: usize, + ) -> Poll> { + assert!(self.supports, "poll_write_file called on an IO without it"); + self.ready = !self.ready; + if !self.ready { + cx.waker().wake_by_ref(); + return Poll::Pending; + } + let mut buf = [0u8; STEP]; + let n = file.read_at(&mut buf[..len.min(STEP)], offset)?; + if n == 0 { + return Poll::Ready(Ok(0)); + } + let written = Pin::new(&mut self.inner).poll_write(cx, &buf[..n]); + if let Poll::Ready(Ok(_)) = written { + self.file_writes.fetch_add(1, Ordering::SeqCst); + } + written + } +} + +/// A file with `FILE` in it, removed on drop. +struct TempFile(PathBuf); + +impl TempFile { + fn new(name: &str) -> (TempFile, Arc) { + let path = + std::env::temp_dir().join(format!("hyper-send-file-{}-{}", std::process::id(), name)); + std::fs::write(&path, FILE).unwrap(); + let file = Arc::new(File::open(&path).unwrap()); + (TempFile(path), file) + } +} + +impl Drop for TempFile { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.0); + } +} + +/// Serve one connection with `respond`, returning the client socket, the +/// count of `poll_write_file` calls, and the connection's result. +async fn serve( + supports: bool, + respond: F, +) -> ( + TcpStream, + Arc, + tokio::task::JoinHandle>, +) +where + F: Fn() -> Response + Send + Sync + 'static, + B: hyper::body::Body + Send + 'static, + B::Data: Send, + B::Error: Into>, +{ + serve_with(http1::Builder::new(), supports, respond).await +} + +async fn serve_with( + builder: http1::Builder, + supports: bool, + respond: F, +) -> ( + TcpStream, + Arc, + tokio::task::JoinHandle>, +) +where + F: Fn() -> Response + Send + Sync + 'static, + B: hyper::body::Body + Send + 'static, + B::Data: Send, + B::Error: Into>, +{ + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let file_writes = Arc::new(AtomicUsize::new(0)); + + let counter = file_writes.clone(); + let respond = Arc::new(respond); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let io = FileIo { + inner: TokioIo::new(socket), + supports, + file_writes: counter, + ready: false, + }; + let service = service_fn(move |_req| { + let respond = respond.clone(); + async move { Ok::<_, Infallible>(respond()) } + }); + builder.serve_connection(io, service).await + }); + + let client = TcpStream::connect(addr).await.unwrap(); + (client, file_writes, server) +} + +/// Read one response: its head, and a body of `Content-Length` bytes. +async fn read_response(client: &mut TcpStream) -> (String, Vec) { + let (head, body, rest) = read_responses(client, Vec::new()).await; + assert!( + rest.is_empty(), + "extra bytes: {:?}", + String::from_utf8_lossy(&rest) + ); + (head, body) +} + +/// Read one response after the bytes in `received`, returning whatever of +/// the next response came with it. +async fn read_responses( + client: &mut TcpStream, + mut received: Vec, +) -> (String, Vec, Vec) { + let mut buf = [0u8; 256]; + let read = timeout(Duration::from_secs(5), async { + loop { + if let Some(end) = find(&received, b"\r\n\r\n") { + let head = String::from_utf8(received[..end].to_vec()).unwrap(); + let len = content_length(&head); + if received.len() >= end + 4 + len { + let body = received[end + 4..end + 4 + len].to_vec(); + let rest = received.split_off(end + 4 + len); + return (head, body, rest); + } + } + let n = client.read(&mut buf).await.unwrap(); + assert!( + n != 0, + "connection closed mid-response: {:?}", + String::from_utf8_lossy(&received) + ); + received.extend_from_slice(&buf[..n]); + } + }) + .await; + read.expect("response never completed") +} + +fn find(haystack: &[u8], needle: &[u8]) -> Option { + haystack.windows(needle.len()).position(|w| w == needle) +} + +fn content_length(head: &str) -> usize { + head.lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + if name.eq_ignore_ascii_case("content-length") { + value.trim().parse().ok() + } else { + None + } + }) + .unwrap_or(0) +} + +fn with_file( + file: &Arc, + offset: u64, + len: u64, + body: &'static [u8], +) -> Response> { + let mut response = Response::new(Full::new(Bytes::from_static(body))); + response + .extensions_mut() + .insert(SendFile::new(file.clone(), offset, len)); + response +} + +const GET: &[u8] = b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"; + +#[tokio::test] +async fn sends_the_file_instead_of_the_body() { + let (_tmp, file) = TempFile::new("instead"); + let (mut client, file_writes, _server) = serve(true, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + client.write_all(GET).await.unwrap(); + let (head, body) = read_response(&mut client).await; + + assert!(head.starts_with("HTTP/1.1 200 OK"), "{head}"); + assert_eq!(content_length(&head), FILE.len()); + assert_eq!(body, FILE); + assert_eq!( + file_writes.load(Ordering::SeqCst), + (FILE.len() + STEP - 1) / STEP + ); +} + +#[tokio::test] +async fn sends_a_range_of_the_file() { + let (_tmp, file) = TempFile::new("range"); + let (mut client, _, _server) = serve(true, move || with_file(&file, 6, 4, b"xxxx")).await; + + client.write_all(GET).await.unwrap(); + let (_, body) = read_response(&mut client).await; + + assert_eq!(body, b"from"); +} + +#[tokio::test] +async fn keeps_the_connection_alive_after_a_file() { + let (_tmp, file) = TempFile::new("keep-alive"); + let (mut client, file_writes, _server) = serve(true, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + for _ in 0..3 { + client.write_all(GET).await.unwrap(); + let (_, body) = read_response(&mut client).await; + assert_eq!(body, FILE); + } + assert_eq!( + file_writes.load(Ordering::SeqCst), + 3 * ((FILE.len() + STEP - 1) / STEP) + ); +} + +#[tokio::test] +async fn finishes_the_file_after_the_client_stops_sending() { + let (_tmp, file) = TempFile::new("half-close"); + // Without half-close, hyper drops a connection whose peer stops sending. + let mut builder = http1::Builder::new(); + builder.half_close(true); + let (mut client, _, _server) = serve_with(builder, true, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + client.write_all(GET).await.unwrap(); + client.shutdown().await.unwrap(); + let (_, body) = read_response(&mut client).await; + assert_eq!(body, FILE); +} + +#[tokio::test] +async fn finishes_the_file_before_closing_a_connection_marked_close() { + let (_tmp, file) = TempFile::new("close"); + let (mut client, _, _server) = serve(true, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + client + .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut received = Vec::new(); + timeout(Duration::from_secs(5), client.read_to_end(&mut received)) + .await + .expect("connection never closed") + .unwrap(); + assert!( + received.ends_with(FILE), + "{:?}", + String::from_utf8_lossy(&received) + ); +} + +#[tokio::test] +async fn sends_pipelined_responses_in_order() { + let (_tmp, file) = TempFile::new("pipelined"); + let (mut client, _, _server) = serve(true, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + client.write_all(&[GET, GET].concat()).await.unwrap(); + let (_, first, rest) = read_responses(&mut client, Vec::new()).await; + let (_, second, rest) = read_responses(&mut client, rest).await; + assert_eq!(first, FILE); + assert_eq!(second, FILE); + assert!(rest.is_empty()); +} + +#[tokio::test] +async fn flushes_the_head_before_the_file_when_pipeline_flush_is_on() { + let (_tmp, file) = TempFile::new("pipeline-flush"); + let mut builder = http1::Builder::new(); + builder.pipeline_flush(true); + let (mut client, _, _server) = serve_with(builder, true, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + client.write_all(&[GET, GET, GET].concat()).await.unwrap(); + let mut rest = Vec::new(); + for _ in 0..3 { + let (head, body, more) = read_responses(&mut client, rest).await; + assert!(head.starts_with("HTTP/1.1 200 OK"), "{head}"); + assert_eq!(body, FILE); + rest = more; + } + assert!(rest.is_empty()); +} + +#[tokio::test] +async fn falls_back_to_the_body_when_the_io_cannot_write_files() { + let (_tmp, file) = TempFile::new("unsupported"); + let (mut client, file_writes, _server) = serve(false, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + client.write_all(GET).await.unwrap(); + let (_, body) = read_response(&mut client).await; + + assert_eq!(body, FALLBACK); + assert_eq!(file_writes.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn falls_back_to_the_body_when_the_length_differs() { + let (_tmp, file) = TempFile::new("length"); + let (mut client, file_writes, _server) = + serve(true, move || with_file(&file, 0, 4, FALLBACK)).await; + + client.write_all(GET).await.unwrap(); + let (_, body) = read_response(&mut client).await; + + assert_eq!(body, FALLBACK); + assert_eq!(file_writes.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn falls_back_to_a_chunked_body() { + let (_tmp, file) = TempFile::new("chunked"); + let (mut client, file_writes, _server) = serve(true, move || { + let chunks = stream::iter(vec![Ok::<_, Infallible>(Frame::data(Bytes::from_static( + FALLBACK, + )))]); + let mut response = Response::new(StreamBody::new(chunks)); + response + .extensions_mut() + .insert(SendFile::new(file.clone(), 0, FILE.len() as u64)); + response + }) + .await; + + client.write_all(GET).await.unwrap(); + let mut received = Vec::new(); + timeout(Duration::from_secs(5), async { + let mut buf = [0u8; 256]; + while !received.ends_with(b"0\r\n\r\n") { + let n = client.read(&mut buf).await.unwrap(); + assert!(n != 0, "connection closed mid-response"); + received.extend_from_slice(&buf[..n]); + } + }) + .await + .expect("response never completed"); + + let received = String::from_utf8(received).unwrap(); + assert!( + received.contains("transfer-encoding: chunked"), + "{received}" + ); + assert!(received.contains("a fallback body"), "{received}"); + assert_eq!(file_writes.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn sends_no_body_for_head() { + let (_tmp, file) = TempFile::new("head"); + let (mut client, file_writes, _server) = serve(true, move || { + with_file(&file, 0, FILE.len() as u64, FALLBACK) + }) + .await; + + client + .write_all(b"HEAD / HTTP/1.1\r\nHost: localhost\r\n\r\n") + .await + .unwrap(); + // A HEAD response announces the length but carries no body, so read the + // head alone and then prove nothing follows it. + let mut received = Vec::new(); + let mut buf = [0u8; 256]; + timeout(Duration::from_secs(5), async { + while find(&received, b"\r\n\r\n").is_none() { + let n = client.read(&mut buf).await.unwrap(); + assert!(n != 0, "connection closed mid-response"); + received.extend_from_slice(&buf[..n]); + } + }) + .await + .expect("response never completed"); + assert!( + received.ends_with(b"\r\n\r\n"), + "body after a HEAD response" + ); + client.write_all(GET).await.unwrap(); + let (_, body) = read_response(&mut client).await; + assert_eq!(body, FILE); + + assert_eq!( + file_writes.load(Ordering::SeqCst), + (FILE.len() + STEP - 1) / STEP + ); +} + +#[tokio::test] +async fn errors_when_the_file_is_shorter_than_its_length() { + let (_tmp, file) = TempFile::new("short"); + // Asks for more of the file than there is, from a body that agrees. + const LONG: &[u8] = b"a fallback body longer than the file"; + let (mut client, _, server) = + serve(true, move || with_file(&file, 0, LONG.len() as u64, LONG)).await; + + client.write_all(GET).await.unwrap(); + let result = timeout(Duration::from_secs(5), server) + .await + .expect("connection never finished") + .unwrap(); + result.expect_err("a short file must fail the connection"); + + // The client gets the file's bytes, and then the connection closes short. + let mut received = Vec::new(); + client.read_to_end(&mut received).await.unwrap(); + assert!( + received.ends_with(FILE), + "{:?}", + String::from_utf8_lossy(&received) + ); +}