base_http/body/

mod.rs

// Released under the MIT License.
// Copyright, 2026, by Samuel Williams.

//! 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>(())
//! ```
//!
//! 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. HTTP/1 needs this before
    /// reading to choose chunked framing, even when the data length is known.
    /// 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.
    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()
    }
}