Module body 

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:

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.

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");

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.

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>(())
})?;

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.

Structs

BodyReader

The readable half of a bounded, single-producer body.

BodyWriter

The unique output half of a BodyReader body.

BufferedBody

An in-memory, replayable sequence of chunks.

EmptyBody

A ready body with no data and a representation length of zero.

HeadBody

An empty body retaining the representation length for a HEAD response.

Enums

BodyError

Failures shared by writable bodies and their producers. Failed reports the producer’s error as its source, so callers can downcast it.

BodyFrame

One owned body event. Initial message fields live in the message head; Headers carries trailing fields, after which no more data is permitted.

Traits

ReadableBody

Protocol-independent body observations and lifecycle.

WritableBody

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.