magmalake flare_mlake

flare_mlake · API · http

flare.http.body

Streaming response body primitives.

The Response.body: List[UInt8] field forces the entire response body to materialise before the first send — a 100MB download allocates 100MB per concurrent client, regardless of the peer's read speed. This module promotes streaming bodies: the reactor pulls chunks on writable edges so that:

  • 100MB downloads cost a few KB of buffer per connection (one chunk in flight at a time).
  • Server-Sent Events, file downloads, log tails, AI-token streaming — all the cases where the response body isn't bounded at handler-return time — work naturally without a separate API surface.
  • Backpressure falls out of construction: when the peer is slow, the kernel send buffer fills, the reactor doesn't ask for the next chunk. No high-water marks needed.

Pieces in place:

  • ChunkSource trait: anything that can produce one chunk of bytes per next(cancel) call.
  • Body trait: anything that can be rendered as a sequence of byte chunks; declares content_length() (Optional, for setting the header) and next_chunk(cancel).
  • InlineBody: today's behaviour packaged as a Body impl — wraps a List[UInt8] and yields it as one chunk.
  • ChunkedBody[Source: ChunkSource]: adapter from a user ChunkSource to the Body trait.

Status (K1 landed):

  • Streaming a Response through the normal Handler path on HTTP/1.1 works: Response.body_stream carries a ChunkSourceBox and the epoll reactor's chunked write loop (flare/http/_reactor/conn_handle.mojo) frames a bounded batch of chunks per writable edge -- the "100MB without 100MB allocation" path. Use stream_response(source) to build one.

Pieces still open:

  • Making Response parametric over B: Body (so handlers can return Response[ChunkedBody[MySource]]). Today's Response keeps its concrete List[UInt8] body plus the optional body_stream; ResponseImpl[B] exists as an additive authoring type, but migrating the hot-path alias is a follow-up.
  • serve_streaming across multiple listeners (multi-worker works; extra bind(List[SocketAddr]) listeners do not).

body_stream itself is served on every wire: HTTP/1.1 chunked, HTTP/2 incremental DATA frames, HTTP/3 over QUIC, and the same h1 chunked framing as ciphertext once the reactor has terminated TLS.

The shape and public API of ChunkSource / Body / InlineBody / ChunkedBody are stable.

Structs

InlineBody

struct InlineBody

Single-shot body — the prior List[UInt8] shape packaged as a Body impl.

On the first call to next_chunk returns the entire byte buffer; on subsequent calls returns None. content_length() is the buffer's length.

Use this when the handler can produce the entire body before returning (the typical request / response flow). For streaming cases reach for ChunkedBody.

Implements Body, Movable

__init__

def __init__(out self, var bytes: List[UInt8])

content_length

def content_length(self) -> Optional[Int]

next_chunk

def next_chunk(mut self, cancel: Cancel) -> Optional[List[UInt8]]

ChunkedBody

struct ChunkedBody[Source: ChunkSource]

Adapts a ChunkSource to the Body trait.

content_length() is always None (chunked sources have no fixed length); the reactor will use Transfer-Encoding: chunked framing when this body is rendered. The Source's next(cancel) is forwarded directly.

Example:

struct Counter(ChunkSource):
    var i: Int
    var buf: String

    def next(mut self, cancel: Cancel) raises -> Optional[List[UInt8]]:
        if cancel.cancelled() or self.i >= 100:
            return Optional[List[UInt8]]()
        self.buf = "data: " + String(self.i) + "\n\n"
        self.i += 1
        var out = List[UInt8]()
        for b in self.buf.as_bytes():
            out.append(b)
        return Optional[List[UInt8]](out^)

Implements Body, Movable

Fields
source: Source

__init__

def __init__(out self, var source: Source)

content_length

def content_length(self) -> Optional[Int]

next_chunk

def next_chunk(mut self, cancel: Cancel) -> Optional[List[UInt8]]

Traits

ChunkSource

A source of byte chunks.

Implementors yield successive chunks via next(cancel). The return has three meanings: a non-empty chunk is data, None is end-of-stream, and an empty chunk means "alive, but nothing right now". The cancel token lets long-running sources short-circuit when the client disconnects (peer FIN), the request times out, or the server drains.

The reactor drains a bounded batch per writable edge rather than a single chunk, so a source that is merely idle must say so with an empty chunk instead of manufacturing filler: an empty chunk ends the batch and the connection retries on the next edge. Returning filler means the reactor emits a whole batch of it per edge.

Implementors may own buffer state across calls (an SSE source might rewrite its internal buffer in place); the chunk returned must remain valid until the next call to next.

Implements Movable

next

def next(mut self, cancel: Cancel) -> Optional[List[UInt8]]

Produce the next chunk, or None to signal end-of-stream.

Args
cancel: Per-request cancel token. Sources should poll cancel.cancelled() between expensive steps.

Raises Error: On unrecoverable source error; the reactor will close the connection.

Body

An HTTP response body.

Two shipped impls today:

  • InlineBody: wraps a List[UInt8]; yields one chunk and signals end-of-stream. The default for -style handlers that return a complete body.
  • ChunkedBody[Source]: adapter around a ChunkSource; yields chunks until the source returns None.

Bodies that know their length up front declare it via content_length() so the reactor can set the Content-Length header and skip Transfer-Encoding: chunked framing. Sources that don't (e.g. SSE) return None from content_length() and the reactor sends chunked.

Implements Movable

content_length

def content_length(self) -> Optional[Int]

Return the body's known total byte count, or None for an unknown / open-ended length.

Bodies that return a concrete Int get Content-Length: N header and inline framing. Bodies that return None get Transfer-Encoding: chunked framing and no Content-Length.

next_chunk

def next_chunk(mut self, cancel: Cancel) -> Optional[List[UInt8]]

Produce the next chunk, or None to signal end-of-stream.

Same three-state contract as ChunkSource.next; the reactor calls this repeatedly within one writable edge, up to its batch budget.

Functions

drain_body

def drain_body[B: Body](mut body: B, cancel: Cancel) -> List[UInt8]

Pull every chunk from body and concatenate into a single List[UInt8].

Test / utility helper for users who want to assert against a body's full contents. Production callers go through the reactor's per-edge pull loop (which lands with the reactor adoption follow-up).

Stops if cancel.cancelled() flips True between chunks.