blackbull.server.connection_protocol

blackbull.server.connection_protocol

Connection transport front end: asyncio.BufferedProtocol over one buffer.

One of these per accepted connection, created before the protocol is known — the shared listener detects HTTP/1.1, h2c, and MQTT off the same resident bytes, so the buffer belongs to the connection, not to any one protocol.

This is what replaces asyncio.StreamReader on the inbound path. The kernel writes straight into the connection's :class:~.read_buffer.ReadBuffer through :meth:ConnectionProtocol.get_buffer, and the actor's coroutine parks on a future that :meth:ConnectionProtocol.buffer_updated resolves.

The actor invariant is intact. One coroutine still owns the connection's state and processes one request at a time; it is woken by the protocol instead of parking inside readuntil. That distinction is the whole design: sanic reaches half our read/parse cost with a parked coroutine too, so the cost was never the coroutine — it was reading through a second buffer.

:class:BufferReader presents the :class:~.recipient.AbstractReader surface so the body recipient and the WebSocket/h2c successors work unchanged, and adds :meth:BufferReader.read_head — the one-scan header read the H/1.1 actor uses instead of a readuntil per line.

Because peeked bytes stay resident, protocol detection can decide without consuming: there is nothing to replay to the winning binding, which is what retires PrefixReader on this path.

BufferReader

Bases: AbstractReader

AbstractReader over a :class:ReadBuffer fed by :class:ConnectionProtocol.

Every method serves from resident bytes first and only parks when it needs more. A pipelined or keep-alive peer's next head is usually already resident, so those reads complete without a loop turn — which is the claim the layered predecessor made and could not deliver, because it sat on a reader that was buffering underneath it.

fill(n) async

Wait until n bytes are resident, consuming nothing.

Free here: resident bytes are already the buffer's normal state, so peeking is just not calling take. It is the reason detection can hand the winning binding this very reader with the stream still whole.

read_head(limit) async

The message head, terminator included — found in one scan.

The override the whole rewrite exists for: the terminator is looked for once, across everything resident, and the head leaves the buffer in a single copy. Resumable, so bytes already scanned are not scanned again when the head arrives split across reads.

Contract as documented on :meth:AbstractReader.read_head — an idle close returns b'' and a truncated one raises with the partial.

ConnectionProtocol

Bases: BufferedProtocol

Buffered-protocol front end for one H/1.1 connection.

buffer property

The connection's buffer, for a successor taking the stream over.

drain() async

Block only while the transport is over its high-water mark.

Returns without awaiting in the common case, which matters: an unconditional await here is one loop turn per response send, the very cost the inbound rewrite is removing on the read side.

linger_close(max_bytes=65536, timeout=0.25) async

Close after briefly discarding whatever the peer is still sending.

Closing a socket with unread bytes in its receive queue makes the kernel send RST, and an RST discards data we already wrote — so a peer that is still mid-send when we answer never sees the response. That is not hypothetical: it is how a 431 for an over-budget header block goes missing, because rejecting at the budget means, by design, not reading the rest.

nginx calls this lingering_close. Both bounds matter: reading without a byte cap hands an attacker the unbounded read the budget exists to refuse, and reading without a deadline lets a slow peer hold the connection open after it has been answered.

Skipped unless we are closing with bytes we chose not to consume. A completed request leaves the buffer empty, so the normal close stays a bare close — lingering on every connection would put a timeout on the teardown path that AsyncioWriter.close deliberately keeps free of even one extra loop turn, for the burst-keepalive workload.

maybe_resume()

Resume reading once the resident bytes fall back to the low mark.

wait_for_data() async

Park until the next arrival, EOF, or connection loss.

One waiter only: a connection is driven by a single actor coroutine, so a second concurrent reader is a bug rather than a case to support.

writelines(parts)

Vectored write — the other half of the send-path size gate.

BaseSender._write_many joins below 32 KiB and comes here above it, so a protocol that offers only :meth:write serves small responses and fails large ones. Delegated to the transport rather than joined here: the selector transport reaches sendmsg(iovec, …) and uvloop does a real vectored write, which is the entire reason the gate has an upper branch.