flare.http.streaming_server

Adopt client into a fresh per-connection handle.

Stable per-connection id (key for handler-owned per-conn state).

Underlying client fd. Fast accessor; does not check state.

Return a Cancel handle bound to this connection's cell.

Flip the connection's cancel cell (framework-internal).

Ask the framework to close this connection after the current lifecycle step returns.

True once request_close has been called.

Attach an upstream source and let the framework run it.

This is the recommended, low-ceremony path: hand the connection the source and the framework takes over -- it reads the fd to watch internally (no descriptors in front code), owns the source for the connection's lifetime (closing it on teardown), and relay_upstream drains it. The front never keeps a per-connection table, never names a file descriptor, and never remembers to close anything.

def on_open(mut self, mut conn: StreamConn) raises:
    conn.attach_upstream(UpstreamChunkSource.connect(self.backend))

Low-level: ask the reactor to watch a raw fd (a front-owned upstream socket / pipe / eventfd) for readability.

Prefer the attach_upstream(UpstreamChunkSource) overload -- it hides the descriptor and the lifetime. Use this only for an upstream that is not an UpstreamChunkSource (a bare pipe, an eventfd). The front then owns fd's lifetime: it opened the upstream and must close it (typically by holding the stream in its own per-connection state and dropping it in on_close). The framework only adds/removes fd from the event loop.

Stop watching the attached upstream and drop any framework-owned source. The reactor unregisters the fd on the next reconcile.

True if an upstream fd is currently attached.

True if a framework-owned UpstreamChunkSource is attached (via the source-taking attach_upstream).

The framework-owned upstream source, for fronts that want to poll it directly instead of using relay_upstream.

Only valid when has_attached_source(); call after attaching via attach_upstream(UpstreamChunkSource).

Drain the attached source into the client with backpressure.

The standard on_upstream body, shipped once so every relay front does not re-derive it: pull ready chunks and send them until the client write buffer hits the high watermark (so a single readable edge cannot overshoot the bound -- the reactor re-arms upstream interest when the client drains), request close on EOF, and stop on pending (the reactor parks on the fd, no spin). No-op if no source is attached.

def on_upstream(mut self, mut conn: StreamConn) raises:
    conn.relay_upstream()

The attached upstream fd, or -1 if none.

Framework-internal: the upstream fd currently registered with the reactor, or -1.

Tune the relay-buffer hi/lo watermarks for this connection.

hi is the occupancy at which the reactor stops reading the attached upstream; lo (< hi) is where it resumes. Values are clamped to a sane order (lo forced below hi) so a front cannot wedge the gate shut. Call in on_open to size the relay pipe before any bytes flow.

The high watermark in bytes.

The low watermark in bytes.

True once unwritten outbound bytes reach the high watermark.

A relay front should check this in its on_upstream drain loop and stop pulling from the upstream when it returns True, so a single readable edge cannot overshoot the bound before the reactor's interest gate takes effect on the next reconcile.

True while upstream reads are gated off by backpressure.

How many times upstream reads were paused (hi crossings).

How many times upstream reads were resumed (lo crossings).

Recompute the upstream read gate from buffer occupancy with hi/lo hysteresis; update the paused flag and crossing counters.

Returns whether the attached upstream fd should currently be read (True = arm INTEREST_READ; False = pause). Pauses on crossing the high watermark, resumes on falling to the low one; between the two it holds the prior state -- the band that stops interest thrash. Framework-internal: the reactor calls this on each reconcile.

Framework-internal: upstream interest bits currently registered (-1 unknown).

Toggle non-blocking mode on the client socket (reactor path sets this once at accept).

Queue data for delivery to the client.

Buffered, not written immediately: the blocking driver flushes with flush_blocking after each lifecycle step; the reactor drains with drain_nonblocking on writable edges. Same call site in both modes, so a front never blocks the event loop.

    Token-burst coalescing: each ``send`` is an O(n) memcpy into
    the single contiguous ``out_buf`` -- it issues no syscall. When a
    front emits K ready chunks in one reactor tick (call ``send`` K
    times), the subsequent drain flushes all K in ONE ``send(2)``,
    removing the per-token syscall round-trip that dominates long-

output nTPOT. The gather is a memcpy into out_buf, not a zero-copy writev over the K chunk buffers; swap in writev_buf over a chunk vector only if that memcpy shows up in a profile -- the syscall count (the nTPOT win) is already 1.

Queue an owned byte list for the client (no Span wrap).

Convenience over send(Span[UInt8, _]) for the common case where a front holds a List[UInt8] -- e.g. a chunk just taken from a ChunkPoll. Same coalescing semantics.

Queue a UTF-8 string for the client (no byte juggling).

Convenience for text protocols (SSE, NDJSON, a status line): conn.send("data: " + line + "\n"). Appends the string's UTF-8 bytes with the same coalescing as send(Span[UInt8, _]).

Queue a Handler-produced Response as this stream's response head (and fixed body, if any).

This is the shared-middleware bridge between serve and serve_streaming: build one Handler middleware stack (e.g. Logger[RequestId[Router]]), call stack.serve(req) inside on_open, then conn.send_response(resp) -- the streaming front gets the exact logging / request-id / routing behaviour the request/response path has, and keeps conn.send(...) for the streamed tail (SSE tokens, relayed chunks). Without this a streaming front had to hand-roll its status line and headers as a raw string, bypassing the middleware stack entirely.

Content-Length is emitted from the body length unless the response already declares Transfer-Encoding or Content-Length (chunked / SSE fronts set Transfer-Encoding and stream the body via send). Connection is always set from keep_alive (any inbound Connection header is dropped). This is HTTP/1.x framing only -- a streaming front speaks h1 to the client; h2/h3 response heads have their own framing on those paths.

The head framing is the shared, transport-agnostic frame_h1_stream_head_into adapter (see flare/http/_server/write.mojo), so an HTTPS front streaming over TlsConnHandle frames its head identically -- the bytes just ride SSL_write instead of send.

Unwritten outbound bytes (len(out_buf) - out_pos).

True if there are unwritten outbound bytes.

Number of send(2) calls issued so far by the drain path. A burst of K chunks queued via send flushes in one syscall, so this advances by ~1 per drain regardless of K.

Reset the write-syscall counter (for microbenches).

Write all pending outbound bytes, blocking. Used by the single-connection driver.

Write as much of the pending buffer as the socket accepts without blocking.

Returns True once the buffer is fully flushed, False if the socket would block (EAGAIN) with bytes still pending -- the caller then keeps INTEREST_WRITE armed and retries on the next writable edge. Raises on a real socket error (EPIPE / ECONNRESET) so the reactor closes the connection.

Append up to max_bytes bytes read from the client onto buf. Returns the count read (0 on peer EOF).

Opt into front-owned inbound body consumption.

Call in on_open for a front that reads a (possibly large) request body itself via read_body. While enabled the reactor no longer drains-and-discards client bytes for FIN detection, so the body bytes reach the front intact; the front pulls them in bounded chunks (peak memory independent of body size) and detects end-of-body when read_body returns eof.

True if the front owns inbound body consumption.

Read up to max_bytes of the inbound request body, without blocking.

The result is a ChunkPoll (so inbound and outbound share one tri-state shape):

• ready(bytes) -- a body chunk is available now;
• eof() -- the client finished the body (clean FIN);
• pending(fd) -- nothing buffered yet; the reactor will fire again on the next readable edge (no busy-poll).

A front loops on this in on_writable (the streaming reactor keeps the connection live), processing each chunk and dropping it, so peak memory is one chunk regardless of total body size up to whatever ceiling the front enforces.

Called once when the connection is accepted. Parse the request, attach an upstream, begin the response.

The one required method: a front that implements nothing here has no way to start a response.

Called when the attached upstream is readable. For a source attached via conn.attach_upstream(UpstreamChunkSource) the whole body is conn.relay_upstream() (drain ready chunks to the client with backpressure, close on EOF). A front that attached a raw fd reads it itself and calls conn.send.

Defaults to doing nothing: a front with no upstream attached never sees this edge.

Called on a writable edge. Emit the next chunk of the response; call conn.request_close() when the stream is complete.

Defaults to doing nothing, which is right for a pure relay: the upstream edge drives the body and the framework handles backpressure.

Called once as the connection is torn down (front FIN, deadline, or request_close). Release per-connection resources.

Defaults to doing nothing: a front whose state is all owned by the framework has nothing to release.