`blackbull.server.recipient`¶

`blackbull.server.recipient` ¶

`AbstractReader` ¶

Bases: ABC

Protocol-agnostic async byte-source.

Mirrors AbstractWriter on the receive side. Implementations wrap a concrete transport so that BaseRecipient subclasses stay runtime-agnostic.

`at_eof()` ¶

Return True once the peer has closed and the buffer is drained.

Default False (callers that need EOF detection — e.g. a long-lived raw-protocol read loop — should use a reader that overrides this).

`readexactly(n)` `async` ¶

Read exactly n bytes. Default: accumulate via :meth:read. Concrete transport readers override this.

`readuntil(sep)` `async` ¶

Read until sep is seen (inclusive). Default: byte-wise via :meth:read. Concrete transport readers override this with the stream's native, buffered implementation.

`AsyncioReader` ¶

Bases: AbstractReader

Adapts an asyncio-compatible stream to AbstractReader.

Accepts any object exposing read(), readuntil(), and readexactly() — the asyncio StreamReader API — so that test doubles such as MagicMock can be injected without ceremony.

`BaseRecipient` ¶

Bases: ABC

Abstract base for ASGI-event receive callables.

__call__ returns an ASGI event dict appropriate to the protocol: - HTTP: {'type': 'http.request', 'body': ..., 'more_body': False} - WebSocket: {'type': 'websocket.connect'}, {'type': 'websocket.receive', ...}, or {'type': 'websocket.disconnect', ...}

The actual byte transport is hidden behind AbstractReader so the recipient logic is decoupled from asyncio internals.

`FragmentAssembler` ¶

Accumulates RFC 6455 fragmented frames and signals message completion.

Feed each data/continuation frame via feed(). Returns (message_opcode, full_payload) when the final FIN=1 continuation arrives; returns None while still accumulating.

Raises ProtocolError on violations: - CONTINUATION frame with no fragmentation in progress (§5.4) - New TEXT/BINARY opener while a fragmented message is open (§5.4)

`feed(opcode, payload, fin, rsv1=False)` ¶

Feed one frame; return (message_opcode, full_payload, compressed) on completion, else None.

`HTTP1Recipient` ¶

Bases: BaseRecipient

Reads an HTTP/1.1 request body and emits a single http.request event.

Body bytes are read lazily on the first __call__ using the Content-Length or Transfer-Encoding header from scope. Subsequent calls return {'type': 'http.disconnect'}.

`drain(max_bytes)` `async` ¶

Discard any unread request body so the next pipelined request parses cleanly. Returns True if fully drained (or the peer disconnected), False if the unread body exceeded max_bytes — the caller should then close the connection rather than keep it alive.

`needs_drain()` ¶

True if a declared request body may still be buffered unread.

A handler that ignores receive (e.g. a 404/405 response to a POST) leaves the body bytes in the reader; the next keep-alive request would then parse them as its request line. The actor consults this after dispatch to decide whether to drain. A body-less request (GET, no Content-Length, not chunked) never needs draining.

`HTTP2Recipient` ¶

Bases: BaseRecipient

Delivers HTTP/2 DATA frames as ASGI http.request events.

The server loop feeds frames via put_DATAFrame() (non-blocking). The ASGI app calls __call__() which suspends until an event is available, hiding the concurrency from both sides.

For GET-style requests (END_STREAM on HEADERS, no DATA frames), the caller invokes :meth:mark_end_of_stream_on_headers instead of pre-queuing an empty http.request event. The Queue is then never allocated — the empty event is synthesized lazily in :meth:__call__ only if the handler reads it.

Consume-based inbound flow control (the Sprint 62 deferral, proposals/consume-based-inbound-flow-control.md): when constructed with a credit_callback, WINDOW_UPDATE credit for a DATA frame is replayed through the callback when the app pops the event — not when the frame is enqueued. A stalled handler then stops crediting, the peer's window closes, and the peer back-pressures instead of overflowing a frame-count queue into RST_STREAM(ENHANCE_YOUR_CALM). In this mode the queue is bounded by credit_budget bytes (the advertised inbound window — a conformant peer cannot exceed it) plus a generous frame-count abuse cap; put_DATAFrame returning False therefore means the peer overran the closed window or dribbled degenerate frames, and the RST is a true abuse backstop. Without a callback the historical bounded-queue, credit-at-enqueue behaviour is preserved (push streams, direct test use).

`credits_on_consume` `property` ¶

True when WINDOW_UPDATE credit is replayed at consume-time.

The actor must then NOT credit at enqueue — the recipient's credit_callback owns the replay.

`mark_end_of_stream_on_headers()` ¶

Mark this stream as ended on HEADERS (no body to deliver).

Replaces put_event({type: http.request, body: b'', more_body: False}) with a flag — saves one asyncio.Queue allocation per body-less request.

`put_DATAFrame(frame)` ¶

Enqueue a DATA frame event. Returns False when the frame must be refused (the caller answers RST_STREAM): queue full in legacy mode; inbound-window overrun or a tiny-frame flood in consume-crediting mode.

`put_disconnect()` ¶

Unblock a waiting call() with an http.disconnect event.

Skipped when end-of-stream-on-headers has been delivered and no queue was ever created — no consumer can be waiting.

`put_event(event)` ¶

Enqueue a pre-built event dict. Returns False if the queue is full.

`take_uncredited()` ¶

Return and clear the un-consumed credit balance.

Bytes enqueued but never popped by the app (a handler that finished — or was RST — without draining its body). The actor replays this to the CONNECTION window when the stream is released, otherwise the shared window leaks shut for every later stream; the stream-level window is moot once the stream closes (RFC 9113 §5.1).

`IncompleteReadError` ¶

Bases: EOFError

Raised by AbstractReader when the peer closes the connection mid-read.

Mirrors asyncio.IncompleteReadError but is not tied to asyncio, so handlers that depend on AbstractReader remain runtime-agnostic.

`PrefixReader` ¶

Bases: AbstractReader

An :class:AbstractReader that replays an already-read prefix.

Connection detection peeks the first bytes of a stream to decide which protocol owns it; wrapping the underlying reader in a PrefixReader hands the still-complete stream to the protocol that claims it — the peeked bytes are served back first, then reads fall through to the underlying reader.

Used by the decouple-connection-detection refactor so the dispatcher no longer consumes protocol-specific bytes on the connection's behalf. The fast native readuntil / readexactly of the underlying reader are used once the prefix is drained, including the seam case where the separator straddles the prefix/underlying boundary.

`ProtocolError` ¶

Bases: Exception

Raised when a WebSocket protocol violation is detected (RFC 6455).

close_code is the RFC 6455 §7.4 status code that should appear in the CLOSE frame sent to the peer. Defaults to 1002 (PROTOCOL_ERROR); UTF-8 violations use 1007.

`RecipientFactory` ¶

Creates the appropriate BaseRecipient for the given protocol.

All methods that need a reader accept a raw asyncio-compatible stream reader and wrap it in AsyncioReader internally.

`WebSocketRecipient` ¶

Bases: BaseRecipient

Reads WebSocket frames and emits ASGI websocket.* events.

First call returns {'type': 'websocket.connect'}. Subsequent calls read the next frame from the transport: - Text frame → {'type': 'websocket.receive', 'text': ..., 'bytes': None} - Binary frame → {'type': 'websocket.receive', 'text': None, 'bytes': ...} - Close frame → {'type': 'websocket.disconnect', 'code': 1000} - Ping frame → sends Pong immediately, then reads the next frame - Pong frame → silently dropped, reads the next frame

Ping/pong handling requires write access to the transport, so the raw writer is stored alongside the reader.

`shutdown()` `async` ¶

Cancel and await the background read-loop task.

Sprint 72 (audit 1.20c) — client sessions call this from close() so no reader task outlives the session (a leaked task warns at event-loop shutdown and keeps reading a dead transport). Idempotent, and safe to call before the first __call__ ever started the loop.

blackbull.server.recipient¶