blackbull.server.sender

blackbull.server.sender

AbstractWriter

Bases: ABC

Protocol-agnostic async byte-sink.

write() is the single responsibility: deliver bytes and ensure they are flushed. Backpressure, buffering, and draining are implementation details of each concrete subclass — callers never call drain() directly.

Implementors wrap a concrete transport (asyncio.StreamWriter, trio MemorySendStream, curio socket, …). BaseSender only depends on this interface, so switching the async runtime requires only a new subclass here.

close() async

Close the underlying transport. Default: no-op.

peer_is_gone()

True when the transport has already recorded the loss.

Asked before a write, because the exception the guard below catches can arrive arbitrarily later: connection_lost is delivered through call_soon, so between the transport recording the loss and the protocol learning of it there is a window in which write() drops silently and drain() returns without raising. Every write in that window is one asyncio counts and warns about.

sendfile(file, offset, count) async

Send up to count bytes from file starting at offset.

Default implementation raises NotImplementedError so callers can detect lack of support and fall back to a read+write loop. Concrete subclasses opt in when the underlying transport supports a zero-copy path (Linux sendfile(2) / loop.sendfile).

Used by the static-file middleware via the http.response.pathsend ASGI extension.

write(data) abstractmethod async

Write data to the transport and ensure it is flushed.

writelines(parts) async

Write multiple byte segments without joining them in user space.

Default joins-and-writes so subclasses can opt out. Override in transports whose writelines does vectored I/O (writev / sendmsg) to skip the full-body memcpy on the static-file cache-hit path.

AsyncioWriter

Bases: AbstractWriter

Adapts an asyncio-compatible stream to AbstractWriter.

The constructor accepts any object that exposes write(bytes) (sync) and drain() (async) — the asyncio StreamWriter API — so that test doubles such as MagicMock can be injected without ceremony.

drain() is called inside write() so the asyncio backpressure mechanism is handled transparently and BaseSender stays runtime-agnostic.

write_timeout (seconds, 0 = disabled) bounds the time spent in drain() waiting for the kernel send buffer to flush. Defends against the slow-read shape of slowloris: a client that reads the response 1 byte/sec fills the send buffer and our drain blocks indefinitely waiting for the peer's TCP window to reopen. On timeout we close the transport and raise ConnectionResetError so the sender treats the failure the same as a peer-side reset.

The bound is carried by the per-process deadline scanner rather than an asyncio timer, because with the timeout enabled every response takes this path — one loop.call_at per write is a per-request cost paid to defend against a case that essentially never happens. The scanner's granularity (BB_DEADLINE_TICK_MS) becomes the slop on when the timeout fires; at the 30 s default that is ~1 %.

sendfile(file, offset, count) async

Zero-copy loop.sendfile against the underlying transport, in bounded chunks.

Raises NotImplementedError (propagated from the loop) when the transport is SSL — TLS framing happens in user-space, so the kernel can't see the plaintext to copy. Callers must catch that and fall back to a read+write loop. Support is a property of the transport, so it is decided on the first chunk: a later chunk cannot discover that sendfile was unavailable all along.

Drains any pending writes first so headers we already buffered precede the file bytes in wire order — under the write bound, like every other drain, so the header flush cannot stall unwatched.

One call per _SENDFILE_CHUNK rather than one for the whole file: each chunk re-arms BB_WRITE_TIMEOUT, which turns "this transfer is stalled" into something expressible without also declaring a legitimately large file to be too slow. Returns the octets actually sent, which is short of count only when the peer stopped accepting.

writelines(parts) async

Vectored write via the underlying StreamWriter.

asyncio.StreamWriter.writelines hands the iterable to transport.writelines, which on the selector transport uses socket.sendmsg(iovec, …) for the immediate-send case and on uvloop is implemented as a real vectored write. Either way the body bytes never get copied into a fresh bytes object before the syscall.

BaseSender

Bases: ABC

Abstract base for ASGI-event → wire-format senders.

__call__ accepts either: - bytes body + optional status and headers: the sender builds and sends the full protocol response (start + body) in one call. - A protocol-specific event dict: dispatched to the appropriate handler.

The actual byte transport is hidden behind AbstractWriter so the sender logic is decoupled from asyncio internals.

mark_client_gone()

The peer is gone — drop further writes instead of raising.

The actor calls this when a read fails in a way that proves the connection is dead (an IncompleteReadError that escaped the body reader), so the response it may still be mid-way through writing dies quietly rather than as a broken-pipe traceback.

This is a control signal between the actor and its sender, which is why it is a method and not an event. As an http.disconnect dict down the send channel it would be the one place the server pushes a receive-side event the wrong way through the pipe, purely because that pipe is already there. The cost is not the dict but the type: every sender's public event union would have to widen to admit a message no application or middleware may ever legally send, and anyone reading the signature would learn the wrong contract.

http.disconnect remains the app-facing spelling on receive(), which is the direction ASGI defines it in.

ConnectionWindow

Shared HTTP/2 connection-level (stream 0) send flow-control window.

One instance per connection, referenced by every stream's :class:HTTP2Sender, so all senders debit and await a single budget.

Without sharing each sender held a private copy of the connection window and debited only that copy, while the actor-level total was only ever incremented — so N concurrent streams could each spend a full 65535-byte window and the server could emit N×65535 bytes with zero real stream-0 credit. A strict peer (nghttp2, grpc-go) treats that as a connection FLOW_CONTROL_ERROR and GOAWAYs (RFC 9113 §6.9.1).

The object is a thin mutable holder: senders read/debit size directly and the owning actor fans out wake-ups to blocked senders on a connection-level WINDOW_UPDATE (it already tracks every live sender).

FlowControlStalled

Bases: Exception

The peer never granted the flow-control credit it was asked for.

Distinct from a write failure: the socket is fine and the peer is answering — it simply declines to accept the response it requested, which is the "data dribble" shape of CVE-2019-9511. Carried as its own type so the stream ends with RST_STREAM(CANCEL) (a stream we gave up on) rather than INTERNAL_ERROR (a server that broke).

HTTP1Sender

Bases: BaseSender

Translates content or ASGI HTTP send events into HTTP/1.1 wire-format bytes.

__call__ accepts two forms:

High-level (bytes body + status): await sender(body_bytes, HTTPStatus.OK, headers=[...]) Writes the status line, headers, blank line, and body in one call.

Low-level (ASGI event dict, for internal/error-handler use): await sender({'type': 'http.response.start', ...}) await sender({'type': 'http.response.body', ...})

http.response.start is buffered until http.response.body arrives so that Content-Length can be injected when the app omits it.

__call__(body, status=HTTPStatus.OK, headers=()) async

Dispatch on body and write the resulting HTTP/1.1 bytes.

Accepted forms:

• bytes — emit a complete response: status line, headers (with Content-Length injected if absent), blank line, body.
• {'type': 'http.response.start', ...} — buffer the status, headers, and trailers flag; nothing is written yet.
• {'type': 'http.response.body', ...} — on the first call after a buffered start, flush the start (adding Content-Length for single-body responses or Transfer-Encoding: chunked when more_body=True); subsequent calls write chunk-framed body bytes and the terminal 0\r\n\r\n when streaming completes without declared trailers.
• {'type': 'http.response.trailers', ...} — write 0\r\n once, followed by trailer fields; the final event adds the empty line.

Unknown event types are logged and dropped; non-dict / non-bytes bodies raise TypeError.

HTTP2Sender

Bases: BaseSender

Translates content or ASGI HTTP send events into HTTP/2 frames.

__call__ accepts four forms:

High-level (bytes body + status): await sender(body_bytes, HTTPStatus.OK, headers=[...]) Sends a HEADERS frame followed by a DATA frame.

Native (:class:~blackbull.native.NativeResponse): await sender(NativeResponse(status=..., header=..., body=...)) One object may carry header, body, and/or trailers; the sender buffers the header arm exactly like the dict start and delegates body/trailers to the shared helpers (HEADERS + DATA [+ trailing HEADERS] coalesce).

Low-level (ASGI event dict): await sender({'type': 'http.response.start', ...}) await sender({'type': 'http.response.body', ...})

Control-plane (raw FrameBase instance): await sender(settings_frame) Serialises and writes the frame directly.

connection_window_size property writable

The shared connection-level send window.

Proxies :attr:ConnectionWindow.size so direct users' per-sender crediting and flow-control tests keep reading/writing sender.connection_window_size while the real state lives on the shared object.

adjust_initial_window(delta)

RFC 9113 §6.9.2 — adjust this sender's stream flow-control window by the change in SETTINGS_INITIAL_WINDOW_SIZE since the peer's last announcement. The window may legitimately become negative.

apply_settings(max_frame_size=None)

Apply SETTINGS parameters that do not require delta tracking.

send_response_headers(status, headers) async

Write a standalone HEADERS frame (END_HEADERS, no END_STREAM) now.

Unlike the http.response.start event — which is buffered until a body event so HEADERS + first DATA can coalesce into one write — this flushes the response HEADERS immediately and leaves the stream open. Required by the RFC 8441 WebSocket-over-HTTP/2 accept: the :status 200 response carries no body, so nothing would ever trigger the deferred flush, and the stream must stay open bidirectionally for the subsequent WebSocket DATA frames.

wake_window()

Wake any blocked _write_data() after a window credit change.

SenderFactory

Creates the appropriate BaseSender for the given protocol.

All methods accept a raw asyncio-compatible stream writer and wrap it in AsyncioWriter internally. To support a different async runtime, implement a new AbstractWriter subclass and pass it directly to the sender constructors instead.

WebSocketSender

Bases: BaseSender

Translates ASGI websocket send events or WebSocketResponse dicts into WebSocket wire frames (RFC 6455).

__call__ accepts an ASGI event dict (as returned by WebSocketResponse): - {'type': 'websocket.send', 'text': ...} → text frame (opcode 0x1) - {'type': 'websocket.send', 'bytes': ...} → binary frame (opcode 0x2) - {'type': 'websocket.close'} → close frame (opcode 0x8) - {'type': 'websocket.accept'} → no-op (handshake already sent)

The status and headers parameters are accepted for interface consistency but are unused for WebSocket connections.

build_response_headers(encoder, stream_id, status, headers, *, end_stream)

Encode a response HEADERS frame (carrying :status) to wire bytes.

Injects a date header when the app did not supply one, mirroring the Headers.save() send path. status may be an HTTPStatus, an int, or a str — it is normalised via str() exactly as the object path does.

build_trailers(encoder, stream_id, headers)

Encode a trailers HEADERS frame (END_HEADERS | END_STREAM, no pseudo-headers) to wire bytes.

This is the basis for the gRPC grpc-status trailers path — a unary RPC response carries a second HEADERS frame with regular fields only.