blackbull.server.recipient

blackbull.server.recipient

Protocol receive channels.

Native next_chunk returns bytes or None; ASGI call returns events. Both share completion state. HTTP/1.1 after_dispatch determines whether to reuse, drain or close; never reuse a framing-invalid stream.

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 n bytes for peek; return False on early EOF.

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.

read_head(limit) async

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

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 → IncompleteReadError carrying the partial, which is a truncated request and not an idle close.

limit bounds the whole head and is what stops an unbounded read; 0 disables it, and an overrun raises ReadLimitExceeded.

The Internals page argues why this is a contract every reader meets rather than a capability the caller sniffs for.

readexactly(n) async

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

readuntil(sep, limit=0) async

Read through sep; limit > 0 bounds it (ReadLimitExceeded).

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

Assemble data frames into (opcode, payload, compressed), or None until complete.

Reject continuation without an opener, a new opener during assembly, and RSV1 on continuations. max_total=0 disables the pre-append size cap; breaches raise MessageTooLarge for CLOSE 1009. This cap bounds compressed bytes; inflation requires its own bound.

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

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

HTTP1Recipient

Bases: BaseRecipient

Yield lazy HTTP/1.1 request-body slices as http.request events.

framing is request_framing(conn.headers), already validated; the body is read by it, not by the headers again. After the terminal body event, subsequent calls return 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, rate or time leaves octets we 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 only place the http.request dict is built, for whoever wants that encoding — a full-form handler calling receive(), or an external host. Connection.body() / stream() take next_chunk and skip the per-chunk dict.

A Content-Length body ends on its last data event; a chunked body ends on a separate empty one.

after_dispatch()

What the connection should do now the handler has answered.

One of CONNECTION_REUSABLE, CONNECTION_NEEDS_DRAIN (an unread body stands between here and the next request), or CONNECTION_MUST_CLOSE.

bind(conn, framing)

Point this recipient at conn, the next request on the connection, whose body framing is framing: (Content-Length or None, chunked) as http1_actor.request_framing returned it.

The reader, chunk size, and deadline are properties of the connection and survive; the per-request state starts over. One recipient per connection is safe only because HTTP/1.1 dispatches one request at a time — not 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 or stalled past body_timeout — 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 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.

None, not b'' — an empty body is a real body. Asking again past the end keeps answering None. A peer that vanishes mid-body raises 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.

The Internals page states the receive-path invariant this implements.

HTTP2Recipient

Bases: BaseRecipient

Deliver HTTP/2 body events from non-blocking put_DATAFrame calls.

With credit_callback, replay flow credit only on consumption; never also credit at enqueue. Bound queued bytes by credit_budget and retain a frame-count abuse cap. A False put refuses the stream. Without the callback, enforce the frame-count queue bound and credit at enqueue. END_STREAM on HEADERS requires mark_end_of_stream_on_headers so the first read delivers an empty final body.

mark_end_of_stream_on_headers()

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

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

Without a queue, a stream that ended on HEADERS has no consumer waiting, so none is created for the disconnect.

put_end_of_stream()

Enqueue an empty final body event when request trailers end the stream.

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

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.

WebSocketRecipient

Bases: BaseRecipient

Receive complete WebSocket messages, after an initial connect event.

require_masked=True receives client frames; False receives server frames. ws_queue_depth=0 reads inline when the app calls receive(); a positive depth uses a bounded read-ahead queue. Both deliver the same events, but inline PING/CLOSE servicing waits for a read or the control watchdog. Registering a websocket_message observer uses the deferred-reader path described in docs/guide/websockets.md; it does not enable eager read-ahead.

terminal_code property

Return the close code once the read side finishes.

__call__() async

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

The only place a websocket.* receive dict is built, for whoever wants that encoding — a raw (conn, receive, send) handler, or an external host. The object form is next_message.

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 WebSocketDisconnect, the same signal 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.

Data frames belong to the app/reader, so skip the servicing coroutine. Observing a control frame also activates the 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 blackbull.websocket.WebSocket.receive publishes.

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

send_touch()

Mark send activity for the idle watchdog.

Does nothing until control frames matter or a listener needs the deferred reader; touch() 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 control-frame processing from buffered bytes.

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 complete frames or headers sufficient to reject them. It never blocks or 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

Disarm the watchdog and cancel the read loop, waiting up to 5 seconds.

Safe to call repeatedly or before reading starts; never joins itself. The caller closes the transport even if the reader does not stop.

start_deferred_reader()

Start the deferred reader task.

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 — a 0-maxsize queue is unbounded, which would drop the backpressure bound.

touch()

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