aioquic

import asyncio import logging from typing import Optional from aioquic.asyncio import connect from aioquic.asyncio.protocol import QuicConnectionProtocol from aioquic.h3.connection import H3_ALPN, H3Connection from aioquic.h3.events import DataReceived, HeadersReceived, H3Event from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import QuicEvent class HttpClientProtocol(QuicConnectionProtocol): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._http = H3Connection(self._quic) self._buffer = b'' self._http_events = asyncio.Queue() self._task = asyncio.create_task(self._http_event_handler()) async def _http_event_handler(self): while True: event = await self._http_events.get() if isinstance(event, HeadersReceived): for k, v in event.headers: print(f"Header: {k.decode()} = {v.decode()}") elif isinstance(event, DataReceived): self._buffer += event.data if event.stream_ended: print(f"Body: {self._buffer.decode()}") self._buffer = b'' def quic_event_received(self, event: QuicEvent) -> None: self._http.handle_event(event) while True: http_event = self._http.pull_http_event() if http_event is None: break self._http_events.put_nowait(http_event) async def get(self, url: str): headers = [ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", url.encode()), (b":path", b"/"), (b"user-agent", b"aioquic-client/1.0"), ] stream_id = self._http.send_headers(stream_id=self._http.get_next_available_stream_id(), headers=headers) self._http.send_data(stream_id=stream_id, data=b'', end_stream=True) self.transmit() async def main(): logging.basicConfig(level=logging.INFO) configuration = QuicConfiguration( is_client=True, alpn_protocols=H3_ALPN, ) # For testing against a local server with a self-signed cert, you might need to disable certificate verification # configuration.verify_mode = ssl.CERT_NONE host = "quic.aiortc.org" port = 4433 async with connect( host=host, port=port, configuration=configuration, create_protocol=HttpClientProtocol, ) as client_protocol: await client_protocol.get(host) await asyncio.sleep(1) # Give time for events to be processed if __name__ == "__main__": asyncio.run(main())