# flare.http.body

> A module of flare_mlake's API.

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`

```mojo
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__`

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

#### `content_length`

```mojo
def content_length(self) -> Optional[Int]
```

#### `next_chunk`

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

### `ChunkedBody`

```mojo
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:
```mojo
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__`

```mojo
def __init__(out self, var source: Source)
```

#### `content_length`

```mojo
def content_length(self) -> Optional[Int]
```

#### `next_chunk`

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

## Traits

### `ChunkSource`

```mojo
undefined
```

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`

```mojo
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`

```mojo
undefined
```

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`

```mojo
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`

```mojo
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`

```mojo
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.