HTTP/2¶

async def serve_http2_connection(
    *,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    request_handler: Callable[[HTTPRequest], Awaitable[HTTPResponse]],
) -> None:
    """Manages an HTTP/2 connection lifecycle.

    Reads frames from the reader, updates connection and stream state machines,
    and dispatches completed requests to the provided handler.

    Args:
        reader (asyncio.StreamReader): The source for incoming frames.
        writer (asyncio.StreamWriter): The destination for outgoing frames.
        request_handler (Callable): A coroutine that processes HTTPRequest
            and returns an HTTPResponse.

    Raises:
        RuntimeError: If the 'h2' library is not available in the environment.
    """
    try:
        h2_config = importlib.import_module("h2.config")
        h2_connection = importlib.import_module("h2.connection")
        h2_events = importlib.import_module("h2.events")
        h2_exceptions = importlib.import_module("h2.exceptions")
    except ImportError as exc:
        raise RuntimeError("HTTP mode 'h2' requires the 'h2' package.") from exc

    h2_configuration_cls = h2_config.H2Configuration
    h2_connection_cls = h2_connection.H2Connection
    data_received_event = h2_events.DataReceived
    request_received_event = h2_events.RequestReceived
    stream_ended_event = h2_events.StreamEnded
    stream_reset_event = h2_events.StreamReset
    h2_error_cls = h2_exceptions.H2Error

    connection = h2_connection_cls(
        config=h2_configuration_cls(client_side=False, header_encoding=None)
    )
    streams: dict[int, _HTTP2StreamState] = {}

    connection.initiate_connection()
    writer.write(connection.data_to_send())
    await writer.drain()

    async def process_stream(stream_id: int) -> None:
        """Finalizes a stream and dispatches it to the request handler.

        Args:
            stream_id (int): The ID of the stream to process.
        """
        stream_state = streams.pop(stream_id, None)
        if stream_state is None:
            return

        body = b"".join(stream_state.body_chunks)
        request = HTTPRequest(
            method=stream_state.method,
            target=stream_state.target,
            http_version="HTTP/2",
            headers=stream_state.headers,
            body=body,
            body_chunks=stream_state.body_chunks or [b""],
        )
        response = await request_handler(request)
        await _send_h2_response(
            connection=connection,
            writer=writer,
            stream_id=stream_id,
            response=response,
        )

    while not writer.is_closing():
        incoming = await reader.read(65_536)
        if not incoming:
            break

        try:
            events = connection.receive_data(incoming)
        except h2_error_cls:
            connection.close_connection()
            writer.write(connection.data_to_send())
            await writer.drain()
            return

        for event in events:
            if isinstance(event, request_received_event):
                method, target, headers = _decode_request_headers(list(event.headers))
                streams[event.stream_id] = _HTTP2StreamState(
                    method=method,
                    target=target,
                    headers=headers,
                )
                if getattr(event, "stream_ended", False):
                    await process_stream(event.stream_id)
            elif isinstance(event, data_received_event):
                stream_state = streams.get(event.stream_id)
                if stream_state is not None:
                    stream_state.body_chunks.append(event.data)
                connection.acknowledge_received_data(
                    event.flow_controlled_length,
                    event.stream_id,
                )
                if getattr(event, "stream_ended", False):
                    await process_stream(event.stream_id)
            elif isinstance(event, stream_ended_event):
                await process_stream(event.stream_id)
            elif isinstance(event, stream_reset_event):
                streams.pop(event.stream_id, None)

        outbound = connection.data_to_send()
        if outbound:
            writer.write(outbound)
            await writer.drain()