base_http/body/

mod.rs

// Released under the MIT License.
// Copyright, 2019-2026, by Samuel Williams.
// Copyright, 2023, by Bruno Sutic.

//! Native HTTP bodies, following `Protocol::HTTP::Body`.
//!
//! [`ReadableBody`] combines synchronous observations and abandonment with asynchronous frame reads using explicit cancellation. I/O-backed implementations retain explicitly constructed scheduler-bound capabilities; asynchronous retirement remains with the resource or driver owner, and [`ReadableBody::close`] signals abandonment without proving retirement. Memory bodies need no scheduler. Readiness, exhaustion, total representation length and replayability have separate meanings.
//!
//! Body length describes the whole representation. It does not decrease as chunks are consumed, and a [`HeadBody`] body may report a nonzero length while emitting no data:
//!
//! ```rust
//! use base_http::body::{BufferedBody, HeadBody, ReadableBody};
//!
//! let mut body = BufferedBody::from(b"hello".to_vec());
//! assert!(body.is_ready());
//! assert_eq!(body.length(), Some(5));
//! assert_eq!(body.read_chunk(), Some(b"hello".to_vec()));
//! assert_eq!(body.length(), Some(5));
//! assert!(body.rewind());
//!
//! let head = HeadBody::for_body(&mut body);
//! assert!(head.is_empty());
//! assert_eq!(head.length(), Some(5));
//! ```
//!
//! Reads move owned [`BodyFrame`] events: `Data { data, end }`, `Headers { fields, end }` and `End`; [`BodyFrame::is_end`] handles both payload-associated and separate completion. Initial fields stay in the message metadata; a body `Headers` frame contains trailers, and no data or further headers may follow it. An empty data payload is ordinary data unless `end` is true. After completion, reads return `End`. `Request::read` and `Response::read` return these frames without mutating the initial fields; forward a frame with [`WritableBody::write_frame`] or collect its fields with `HttpHeaders::append_trailers`. [`BodyFrame::map_data`] changes the data owner while preserving fields and completion, so forwarding does not require a common allocation, cloning, boxing or copying payload bytes.
//!
//! [`BodyReader`] and [`BodyWriter`] are bounded consumer/producer halves implementing [`ReadableBody`] and [`WritableBody`]. [`BodyReader::bounded`] applies backpressure under both a byte and a frame limit. Normal write closure drains accepted frames and marks the last queued payload as final; failure discards queued data and preserves the cause.
//!
//! ```rust
//! use base_http::body::{BodyFrame, BodyReader, BodyError, ReadableBody};
//! use base_scheduler::Cancellation;
//! use futures_lite::future::{block_on, zip};
//!
//! let (mut reader, mut writer) = BodyReader::bounded(1024, 16);
//! let producer = async {
//!     writer.write_frame(
//!         BodyFrame::Data { data: b"hello".to_vec(), end: true },
//!         &Cancellation::never(),
//!     ).await?;
//!     Ok::<_, BodyError>(())
//! };
//! let consumer = async {
//!     let mut bytes = Vec::new();
//!     loop {
//!         let frame = reader.read(&Cancellation::never()).await?;
//!         let end = frame.is_end();
//!         match frame {
//!             BodyFrame::Data { data, .. } => bytes.extend(data),
//!             BodyFrame::Headers { fields, .. } => {
//!                 // Forward fields, inspect them, or explicitly collect them.
//!                 assert!(fields.is_empty());
//!             }
//!             BodyFrame::End => {}
//!         }
//!         if end { break; }
//!     }
//!     Ok::<_, BodyError>(bytes)
//! };
//! let (sent, received) = block_on(zip(producer, consumer));
//! sent?;
//! assert_eq!(received?, b"hello");
//! # Ok::<(), BodyError>(())
//! ```
//!
//! # Polling reads alongside other I/O
//!
//! A streaming consumer, such as a server-sent events reader, often waits for the next frame and for another source, such as a command channel or a timer, at the same time. The [`ReadableBody::read`] contract requires that dropping a pending read loses no frame, so create a new read for each wait and drop it when the other source wins. The next read continues at the same position. I/O-backed bodies meet this when their underlying reader is cancel-safe, as scheduler sockets and TLS streams are.
//!
//! The read future is not `Unpin`, and macros such as `futures_util::select!` require every arm to be `FusedFuture + Unpin`. Pin each read with [`std::pin::pin!`] and fuse it with `FutureExt::fuse`, and do the same for other futures which are not already fused, such as a channel's `recv()`; no custom wrapper type or allocation is needed. Combinators such as `futures_lite::future::or`, whose futures produce one output type, need neither. When the other source only needs to stop reading, a [`Cancellation`] passed to `read` is simpler still.
//!
//! ```rust
//! use base_http::body::{BodyError, BodyFrame, BodyReader, ReadableBody};
//! use base_scheduler::Cancellation;
//! use futures_lite::future::block_on;
//! use futures_util::{FutureExt, StreamExt, select_biased};
//! use std::pin::pin;
//!
//! enum Command {
//!     Stop,
//! }
//!
//! let (mut body, mut writer) = BodyReader::bounded(1024, 16);
//! let (sender, mut commands) = futures_channel::mpsc::unbounded();
//! let never = Cancellation::never();
//! block_on(async {
//!     writer.write_frame(BodyFrame::Data { data: b"event".to_vec(), end: false }, &never).await?;
//!     let mut received = Vec::new();
//!     loop {
//!         // Each iteration races a new read, which is dropped while still waiting when the command wins.
//!         let mut read = pin!(body.read(&never).fuse());
//!         select_biased! {
//!             frame = read => {
//!                 let frame = frame?;
//!                 // Check completion before moving the frame's contents.
//!                 let end = frame.is_end();
//!                 if let BodyFrame::Data { data, .. } = frame {
//!                     received.extend(data);
//!                 }
//!                 if end {
//!                     break;
//!                 }
//!                 sender.unbounded_send(Command::Stop).unwrap();
//!             }
//!             command = commands.next() => match command {
//!                 Some(Command::Stop) | None => break,
//!             },
//!         }
//!     }
//!     assert_eq!(received, b"event");
//!
//!     // The dropped read consumed nothing, so the body is still read in order.
//!     writer.write_frame(BodyFrame::Data { data: b"more".to_vec(), end: true }, &never).await?;
//!     let BodyFrame::Data { data, end } = body.read(&never).await? else {
//!         unreachable!();
//!     };
//!     assert_eq!((data.as_slice(), end), (&b"more"[..], true));
//!     Ok::<_, BodyError>(())
//! })?;
//! # Ok::<(), BodyError>(())
//! ```
//!
//! There is one pull-body abstraction. A protocol driver reads bodies directly in its existing future, so bodies do not spawn tasks. Custom bodies can read incoming input while producing outgoing chunks. Imperative producers can use the bounded pair with explicitly supervised concurrent work.
use base_scheduler::Cancellation;
use std::future::Future;

mod buffered;
mod empty;
mod error;
mod frame;
mod head;
mod writable;

pub use buffered::BufferedBody;
pub use empty::EmptyBody;
pub use error::BodyError;
pub use frame::BodyFrame;
pub use head::HeadBody;
pub use writable::{BodyReader, BodyWriter};

/// Protocol-independent body observations and lifecycle.
///
/// No observation performs I/O or requires a scheduler. Readiness, exhaustion,
/// known representation length and rewindability are independent properties.
pub trait ReadableBody: Send {
    /// Application data, without imposing completion-I/O buffer requirements.
    type Data: AsRef<[u8]> + Send;
    type Error: std::error::Error + Send + Sync + 'static;

    /// Whether another read can proceed without waiting for produced data.
    ///
    /// This is a conservative batching hint. Terminal completion/error can also
    /// be ready. It does not promise that an I/O operation will finish in one poll.
    fn is_ready(&self) -> bool {
        false
    }

    /// Whether no unread payload frames remain. False can mean unknown; an open
    /// queue with no frames is not exhausted. This does not suppress read errors.
    fn is_empty(&self) -> bool {
        false
    }

    /// Total representation length, independent of the current read position.
    fn length(&self) -> Option<u64> {
        None
    }

    /// Whether this body can emit trailing headers. An HTTP/1.1 response needs
    /// this before reading to choose chunked framing, even when the data length
    /// is known. An HTTP/1 request of known length uses `Content-Length` and
    /// drops any trailers whatever this reports. Returning false promises that
    /// no Headers frame will be produced.
    fn may_have_trailers(&self) -> bool {
        true
    }

    fn is_rewindable(&self) -> bool {
        false
    }

    /// Restore the read position when supported, reporting success.
    fn rewind(&mut self) -> bool {
        false
    }

    /// Abandon further application reads, optionally reporting their failure.
    ///
    /// This synchronous operation does not guarantee asynchronous resource
    /// retirement. An I/O-backed implementation must leave cleanup with an
    /// explicit resource/driver owner. Cancelling or dropping a read does not
    /// prove that backend access ended, nor that a connection can be reused.
    fn close(&mut self, error: Option<Self::Error>);

    /// Read one owned frame. Pre-cancellation prevents consumption. The caller
    /// stops after `frame.is_end()`; subsequent reads return `BodyFrame::End`.
    /// Headers occur at most once, after all data. Dropping a pending read is
    /// not transport retirement, and must not lose an unreturned queued frame.
    ///
    /// Callers can therefore race a read against other futures; see [polling reads alongside other I/O](self#polling-reads-alongside-other-io).
    fn read(
        &mut self,
        cancellation: &Cancellation,
    ) -> impl Future<Output = Result<BodyFrame<Self::Data>, Self::Error>> + Send;
}
impl<R: ReadableBody + ?Sized> ReadableBody for &mut R {
    type Data = R::Data;
    type Error = R::Error;

    fn is_ready(&self) -> bool {
        (**self).is_ready()
    }
    fn is_empty(&self) -> bool {
        (**self).is_empty()
    }
    fn length(&self) -> Option<u64> {
        (**self).length()
    }
    fn may_have_trailers(&self) -> bool {
        (**self).may_have_trailers()
    }
    fn is_rewindable(&self) -> bool {
        (**self).is_rewindable()
    }
    fn rewind(&mut self) -> bool {
        (**self).rewind()
    }
    fn close(&mut self, error: Option<Self::Error>) {
        (**self).close(error)
    }

    async fn read(
        &mut self,
        cancellation: &Cancellation,
    ) -> Result<BodyFrame<Self::Data>, Self::Error> {
        (**self).read(cancellation).await
    }
}

/// A body producer accepting owned frames with backpressure.
/// Normal closure preserves accepted frames; failure abandons unfinished output.
/// Flushing exposes accepted frames to the consumer, without promising peer receipt.
pub trait WritableBody: Send {
    type Data: AsRef<[u8]> + Send;
    type Error: std::error::Error + Send + Sync + 'static;

    /// Accept one whole frame. Headers follow all data; a terminal frame closes
    /// output atomically with acceptance. Later writes fail. Queue capacity
    /// becoming available does not mean the transport has released its buffers.
    fn write_frame(
        &mut self,
        frame: BodyFrame<Self::Data>,
        cancellation: &Cancellation,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;

    /// Write a nonfinal data frame. Use `write_frame` to carry fields or end.
    fn write(
        &mut self,
        data: Self::Data,
        cancellation: &Cancellation,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        self.write_frame(BodyFrame::Data { data, end: false }, cancellation)
    }
    fn flush(
        &mut self,
        cancellation: &Cancellation,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
    fn close_write(&mut self, error: Option<Self::Error>);
    fn abort(&mut self);
}

impl<W: WritableBody + ?Sized> WritableBody for &mut W {
    type Data = W::Data;
    type Error = W::Error;
    async fn write_frame(
        &mut self,
        frame: BodyFrame<Self::Data>,
        cancellation: &Cancellation,
    ) -> Result<(), Self::Error> {
        (**self).write_frame(frame, cancellation).await
    }
    async fn write(
        &mut self,
        data: Self::Data,
        cancellation: &Cancellation,
    ) -> Result<(), Self::Error> {
        (**self).write(data, cancellation).await
    }
    async fn flush(&mut self, cancellation: &Cancellation) -> Result<(), Self::Error> {
        (**self).flush(cancellation).await
    }
    fn close_write(&mut self, error: Option<Self::Error>) {
        (**self).close_write(error)
    }
    fn abort(&mut self) {
        (**self).abort()
    }
}