HTTP/3¶

async def create_http3_server(
    *,
    config: PalfreyConfig,
    request_handler: Callable[[HTTPRequest, ClientAddress, ServerAddress], Awaitable[HTTPResponse]],
) -> Any:
    """
    Create and start an HTTP/3 QUIC listener.

    Args:
        config (PalfreyConfig): Runtime configuration.
        request_handler (Callable[[HTTPRequest, ClientAddress, ServerAddress], Awaitable[HTTPResponse]]):
            Coroutine that receives parsed HTTP/3 requests and returns a response.

    Returns:
        Any: The aioquic server object with `close()` and `wait_closed()` methods.

    Raises:
        RuntimeError: If required dependencies or TLS files are unavailable.
    """
    try:
        aioquic_asyncio = importlib.import_module("aioquic.asyncio")
        aioquic_asyncio_protocol = importlib.import_module("aioquic.asyncio.protocol")
        aioquic_h3_connection = importlib.import_module("aioquic.h3.connection")
        aioquic_h3_events = importlib.import_module("aioquic.h3.events")
        aioquic_quic_configuration = importlib.import_module("aioquic.quic.configuration")
        aioquic_quic_events = importlib.import_module("aioquic.quic.events")
    except ImportError as exc:
        raise RuntimeError("HTTP mode 'h3' requires the 'aioquic' package.") from exc

    serve = aioquic_asyncio.serve
    quic_connection_protocol_cls = aioquic_asyncio_protocol.QuicConnectionProtocol
    h3_alpn = aioquic_h3_connection.H3_ALPN
    h3_connection_cls = aioquic_h3_connection.H3Connection
    data_received_event = aioquic_h3_events.DataReceived
    headers_received_event = aioquic_h3_events.HeadersReceived
    quic_configuration_cls = aioquic_quic_configuration.QuicConfiguration
    protocol_negotiated_event = aioquic_quic_events.ProtocolNegotiated

    if not config.ssl_certfile:
        raise RuntimeError("HTTP mode 'h3' requires --ssl-certfile.")
    if not config.ssl_keyfile:
        raise RuntimeError("HTTP mode 'h3' requires --ssl-keyfile.")

    quic_configuration = quic_configuration_cls(is_client=False, alpn_protocols=h3_alpn)
    quic_configuration.load_cert_chain(
        certfile=config.ssl_certfile,
        keyfile=config.ssl_keyfile,
        password=config.ssl_keyfile_password,
    )

    class _PalfreyHTTP3Protocol(quic_connection_protocol_cls):
        """QUIC protocol adapter that translates HTTP/3 frames into ASGI requests.

        This protocol handler manages the lifecycle of a QUIC connection,
        negotiating HTTP/3 via ALPN and dispatching stream events to
        the underlying H3 connection state machine.
        """

        def __init__(self, *args: Any, **kwargs: Any) -> None:
            """Initializes the HTTP/3 protocol state.

            Args:
                *args: Positional arguments for the base QuicConnectionProtocol.
                **kwargs: Keyword arguments for the base QuicConnectionProtocol.
            """
            super().__init__(*args, **kwargs)
            self._http: Any | None = None
            self._streams: dict[int, _HTTP3StreamState] = {}
            self._tasks: set[asyncio.Task[None]] = set()

        def connection_lost(self, exc: Exception | None) -> None:
            """Cleans up active tasks when the QUIC connection is closed.

            Args:
                exc (Exception | None): The reason for connection loss, if any.
            """
            for task in list(self._tasks):
                task.cancel()
            super().connection_lost(exc)

        def quic_event_received(self, event: Any) -> None:
            """Handles incoming QUIC events and routes them to HTTP/3 logic.

            Args:
                event (Any): The QUIC event (e.g., DataReceived, ProtocolNegotiated).
            """
            if isinstance(event, protocol_negotiated_event) and event.alpn_protocol in h3_alpn:
                self._http = h3_connection_cls(self._quic)

            if self._http is None:
                return

            for http_event in self._http.handle_event(event):
                if isinstance(http_event, headers_received_event):
                    method, target, headers = _decode_request_headers(list(http_event.headers))
                    self._streams[http_event.stream_id] = _HTTP3StreamState(
                        method=method,
                        target=target,
                        headers=headers,
                    )
                    if http_event.stream_ended:
                        self._schedule_request(http_event.stream_id)
                elif isinstance(http_event, data_received_event):
                    stream_state = self._streams.get(http_event.stream_id)
                    if stream_state is not None:
                        stream_state.body_chunks.append(http_event.data)
                    if http_event.stream_ended:
                        self._schedule_request(http_event.stream_id)

            self.transmit()

        def _schedule_request(self, stream_id: int) -> None:
            """Schedules the asynchronous dispatch of a completed request.

            Args:
                stream_id (int): The ID of the completed stream.
            """
            task = asyncio.create_task(self._dispatch_request(stream_id))
            self._tasks.add(task)
            task.add_done_callback(self._tasks.discard)

        async def _dispatch_request(self, stream_id: int) -> None:
            """Processes a completed stream and sends the response.

            Extracts request data, calls the request handler, and serializes
            the resulting HTTPResponse back to the QUIC stream.

            Args:
                stream_id (int): The ID of the stream to dispatch.
            """
            if self._http is None:
                return

            stream_state = self._streams.pop(stream_id, None)
            if stream_state is None:
                return

            transport = getattr(self, "_transport", None)
            client = _normalize_address(
                transport.get_extra_info("peername") if transport is not None else None,
                default_host="0.0.0.0",
                default_port=0,
            )
            server = _normalize_address(
                transport.get_extra_info("sockname") if transport is not None else None,
                default_host=config.host,
                default_port=config.port,
            )

            body = b"".join(stream_state.body_chunks)
            request = HTTPRequest(
                method=stream_state.method,
                target=stream_state.target,
                http_version="HTTP/3",
                headers=stream_state.headers,
                body=body,
                body_chunks=stream_state.body_chunks or [b""],
            )

            try:
                response = await request_handler(request, client, server)
            except Exception:
                response = HTTPResponse(
                    status=500,
                    headers=[(b"content-type", b"text/plain")],
                    body_chunks=[b"Internal Server Error"],
                )

            headers, payload = _encode_response_headers(response)
            self._http.send_headers(stream_id=stream_id, headers=headers, end_stream=not payload)
            if payload:
                self._http.send_data(stream_id=stream_id, data=payload, end_stream=True)
            self.transmit()

    return await serve(
        config.host,
        config.port,
        configuration=quic_configuration,
        create_protocol=_PalfreyHTTP3Protocol,
    )