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.

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.

sendfile(file, offset, count) async

Zero-copy loop.sendfile against the underlying transport.

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.

Drains any pending writes first so headers we already buffered precede the file bytes in wire order.

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.

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 (bug 1.2) 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).

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.
• {'type': 'http.response.trailers', ...} — write the terminal 0\r\n followed by the trailer headers (chunked encoding).

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 three forms:

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

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 (bug 1.2).

Proxies :attr:ConnectionWindow.size so existing call sites — the experimental client's 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.