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).

buffered_len()

Bytes currently buffered, unconsumed. Default 0.

fill(n) async

Buffer up to n bytes for peeking. False if EOF came first.

This is what lets connection detection choose a protocol without eating the bytes it inspected. A reader that owns its buffer overrides this and genuinely consumes nothing.

The default cannot: it has only :meth:read, so it does consume, and parks what it took in :attr:_ahead. :meth:take_ahead hands that back to the caller, which restores the stream by wrapping this reader in a :class:PrefixReader. Same outcome, one indirection more — and it keeps every reader, including test doubles, usable for detection without each one reimplementing pushback.

has_buffered()

True when bytes are already buffered, so a read need not block.

The WebSocket control-frame watchdog probes this to decide whether it can service a frame without blocking the caller's path. Default False — a reader that cannot report buffering is treated as "nothing available", which only disables proactive servicing, never correctness (control frames are still serviced at the next read).

peek(n)

Up to n buffered bytes without consuming them.

Returns whatever the default :meth:fill parked, so detection can inspect it; a reader that owns its buffer overrides this to read straight out of it.

read_head(limit) async

One message head — start line, field lines, terminator included.

Part of the reader contract rather than something the caller sniffs for, so a protocol asks for a head the same way whatever is underneath it. A reader that owns its buffer overrides this to find the terminator in a single scan and to return without a loop turn when the head is already resident; the default below is what a reader with only readuntil can do — one call per line.

Three outcomes, because the caller answers each differently:

• a complete head → returned;
• EOF before a single byte of it → b'', an idle close;
• EOF part-way through → :class:IncompleteReadError carrying the partial, which is a truncated request and not an idle close.

limit bounds the whole head (0 disables it). Passing it is what stops an unbounded read; :class:ReadLimitExceeded says the budget was passed and carries the bytes, and the protocol decides which status that becomes.

readexactly(n) async

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

readuntil(sep, limit=0) async

Read until sep is seen, choosing one limit policy at entry.

take_ahead()

Bytes this reader consumed while filling, cleared.

Empty for a reader whose :meth:fill truly peeks — which is the whole point: the caller wraps only when there is something to replay.

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.

Pass-through by design: every method delegates to the stream's own native, buffered implementation with nothing layered on top. Detection's pushback is the base class's :attr:~AbstractReader._ahead / :class:PrefixReader pair — one mechanism for every reader that cannot truly peek, rather than a private copy here. Only the buffer-inspecting probes below know they are sitting on a StreamReader.

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)

max_total bounds the reassembled message; 0 disables it. The check runs before the append, so the frame that crosses the bound is refused rather than accumulated and then regretted — a bound enforced after the fact would have already paid for the attack. Raises :class:MessageTooLarge, which the caller turns into CLOSE 1009. Note this bounds the compressed bytes when permessage-deflate is in play; the inflated size is bounded separately, because only one of the two is knowable here.

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 of the :class:Connection it is bound to. Subsequent calls return {'type': 'http.disconnect'}.

must_close property

This connection cannot carry another request.

Two causes, one consequence. A chunked-framing violation leaves the byte stream desynced; a body refused for size leaves octets we deliberately did not read. Either way the bytes that follow are the peer's to choose, and parsing them as the next request line is the request-smuggling shape — so the answer is to close, not to resynchronise.

__call__() async

The ASGI receive channel: the same body, encoded as event dicts.

The compat surface, and the only place the http.request dict is built. It costs one dict per chunk and is paid for by the caller that wanted the ASGI encoding — a full-form handler calling receive(), or an external host. Connection.body() / stream() take :meth:next_chunk and pay nothing.

The event sequence is unchanged: more_body is recovered from _done, which :meth:next_chunk has just set, so a Content-Length body still ends on its last data event while a chunked body still ends on a separate empty one.

after_dispatch()

What the connection should do now the handler has answered.

One question, because it is one judgement. Asking must_close and needs_drain() separately and combining them puts the verdict in the caller and leaves the two predicates free to drift apart — the recipient is the object that knows whether the message boundary survived, so it should say what follows from that.

Also one call per request instead of two on the keep-alive path.

bind(conn)

Point this recipient at conn, the next request on the connection.

The reader, chunk size, and deadline are properties of the connection and survive; the framing state is re-derived from the new head. One recipient per connection instead of one per request is the same trade the sender already makes — safe for HTTP/1.1 because a connection dispatches one request at a time, and not safe for HTTP/2, whose streams are concurrent.

The split is the whole contract: any per-request field left out of this method would leak from request N into request N+1, so new state belongs here, not in __init__.

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. A body-less request (GET, no Content-Length, not chunked) never needs draining.

Kept as the named question it is, for tests and for a directly-driven recipient; the actor asks :meth:after_dispatch instead, which answers this and must_close in one call.

next_chunk() async

The next body chunk, or None once the body is complete.

The native receive channel: a chunk is the bytes themselves, and the end of the body is carried by the call protocol rather than by a field beside the payload. more_body was never information about the chunk — it is the channel's state — and every internal consumer did the same one thing with it (if not more_body: break), so the boundary belongs where a Python caller already looks for it.

None, not b'': an empty body is a real body, the same reason :class:~blackbull.native.NativeResponse decides presence with is not None. On both framings the sentinel is unambiguous — a zero-length chunk is the terminator in chunked encoding (RFC 9112 §7.1), and a Content-Length slice is never empty.

Asking again past the end keeps answering None. A peer that vanishes mid-body raises :class:ClientDisconnected — a truncated upload must never read as a complete one — and so does a body-read timeout, which is recorded as a cap hit first.

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: 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 queue is bounded and credit is issued at enqueue instead (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.

make_item(frame) staticmethod

The queue's payload: (chunk, end_of_stream).

The pair the two channels need, and nothing else — __call__ re-encodes it as an ASGI event, :meth:next_chunk hands the bytes straight over. Building the dict here charged every H2 body reader for the encoding, including the ones that never read it.

end_stream is coerced: the frame carries the raw flag bit (DataFrameFlags.END_STREAM & flags, an int), and the queue item is a value both channels read directly, so it holds the answer rather than the wire encoding of it.

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.

next_chunk() async

The next body chunk, or None once the stream has ended.

The H2 half of the native receive channel — same contract as :meth:HTTP1Recipient.next_chunk, so Connection.body() / stream() read one protocol and get both.

put_DATAFrame(frame)

Enqueue a DATA frame event. Returns False when the frame must be refused (the caller answers RST_STREAM): queue full when no credit callback is installed; inbound-window overrun, a tiny-frame flood, or a body limit (BB_MAX_BODY_SIZE / BB_MIN_BODY_RATE) 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_end_of_stream()

Enqueue a clean, empty end-of-body.

The trailers case (RFC 9113 §8.1): a second HEADERS on an open request stream ends the body without carrying any. Enqueues the native pair: building an http.request dict here only to translate it back one line later would put a request-dict producer back on the native path.

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.

partial property

Whatever had been read when the peer went away.

A truncated head and an idle close are the same exception with different payloads, and the caller answers 400 for one and nothing at all for the other — so the payload is part of the contract, not a debugging aid.

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.

This is what keeps the dispatcher from consuming 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.

ReadLimitExceeded

Bases: Exception

A bounded reader operation was given a byte budget and passed it.

Belongs to the reader contract rather than to any protocol: the reader is told a budget and reports that it was passed. Which status that becomes (431 for a head with too many fields, 400 for bytes that were never a head at all) is the protocol's decision — so the reader hands back what it :attr:saw, and every reader answers that question off the same evidence.

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.

Client callers inject the ownership names 'client_ws_max_frame_payload', and 'client_ws_max_message_size', at these shared rejection sites.

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.

Two read modes, selected by ws_queue_depth.

0 (default) — inline. Frames are read in the app's own task, only when it calls receive(). There is no background task and no queue, so a message costs no handoff. This is the difference between WebSocket's 4.09 loop touches/req and HTTP/1.1's 2.06: read-ahead is exactly one extra future plus one extra call_soon per message.

> 0 — eager. A background task reads ahead into a bounded queue of that depth. Costs the handoff, and buys read-ahead: control frames are serviced while the handler is busy, so a PING is answered even between receive() calls, and up to depth messages buffer under a slow app.

Both modes deliver an identical ASGI event sequence to the app; only the timing of control-frame servicing and the existence of buffering differ. Inline mode still answers PING and echoes CLOSE per RFC 6455 §5.5 — it does so when the app drives the next read. RFC 6455 §5.5.2 permits a delayed PONG, which is what makes inline mode conformant.

The one thing that can tell the modes apart is the websocket_message Level B event, which fires when the server reads a message rather than when the app consumes it — a handler that never calls receive() must still produce it. A registered listener does not force read-ahead on, though: a consuming handler is already reading, so the reader is only marked deferred and the idle watchdog starts it if the handler goes quiet.

terminal_code property

The RFC 6455 §7.4 close code, once the read side has finished.

The single record of how this connection ended. The actor keeps no copy of its own: that would mean intercepting every event to look for a disconnect, and two records of one fact is one place for them to disagree.

__call__() async

The ASGI receive channel: the same messages, encoded as dicts.

The compat surface, and the only place a websocket.* receive dict is built — minted per call for whoever wants that encoding: a raw (conn, receive, send) handler, or an external host. The object form takes :meth:next_message and pays nothing.

await_connect() async

Consume the opening handshake on the native channel.

The raw (conn, receive, send) form reads a websocket.connect dict for this; the object form has no use for the envelope, so the native channel just records that the handshake was taken. A peer that gave up mid-handshake raises :class:WebSocketDisconnect, the same signal :meth:next_message gives.

has_buffered()

True when inbound bytes are already buffered (a read won't block).

The actor's send-time servicing probes this so it skips the servicing call entirely on the common empty-buffer path.

has_control_frames_buffered()

True when a control frame leads the inbound buffer.

Synchronous, O(1) gate for send-time servicing: with only data frames buffered (a flood), the servicing coroutine's flag churn and _frame_bytes_needed scan would run per message for nothing — a data frame is owned by the app/reader, not the servicer. Also marks the connection as having observed a control frame, which activates the per-message watchdog work.

next_message() async

The next complete application message: str text, bytes binary.

The native receive channel. Fragments are already reassembled (RFC 6455 §5.4), so what comes back is always a whole message, and the Python type is the text/binary discriminator — the same contract :meth:blackbull.websocket.WebSocket.receive publishes.

Raises :class:~blackbull.websocket.WebSocketDisconnect when the peer closes, carrying the RFC 6455 §7.4 status code, and re-raises a :class:ProtocolError the read side recorded.

send_touch()

Mark send activity for the idle watchdog, at one bool's cost.

The watchdog is armed once at connect (an idle connection with a buffered control frame must still be serviced even if it never touches); this only keeps the deadline fresh once control frames matter or a listener needs the deferred reader. touch() itself re-arms a missing watchdog, so a send before the connect receive is still safe. There is deliberately no send-time servicing fast path: the watchdog alone bounds PONG latency to ~one scanner tick, which is the documented contract.

service_available_control_frames() async

Non-blocking servicing of fully-buffered inbound control frames.

Answers PINGs and echoes CLOSE that arrived while the handler was doing something other than receive() (send-time servicing) or after it went quiet (the idle watchdog). Reads only frames already fully buffered, so it never blocks and never steals the wire from an inline receive() (guarded by _reading / _servicing). A data frame stops the loop without being consumed — the app or a reader owns it. Returns True if any frame was serviced.

shutdown() async

Cancel and await the background read-loop task, and disarm the idle watchdog.

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.

start_deferred_reader()

Start the deferred reader task.

Called by the idle watchdog once the app has stopped driving receive() on a connection that needs read-ahead (a websocket_message listener). Idempotent and safe: refuses while a reader already owns the wire, while the app is mid-read, or after the read side terminated.

A listener can need read-ahead with the depth left at 0, so the queue falls back to the standard depth rather than a 0-maxsize (i.e. unbounded) one, which would drop the backpressure bound.

touch()

Mark connection activity (receive or send) for the idle watchdog.

The default hot path pays one loop.time() + a comparison per message; an actively-driven connection never fires the watchdog.