### Run HTTP/3 server example Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Start the example HTTP/3 server which supports both HTTP/0.9 and HTTP/3 protocols. Requires certificate and private key files. ```console python examples/http3_server.py --certificate tests/ssl_cert.pem --private-key tests/ssl_key.pem ``` -------------------------------- ### Run DNS over QUIC server Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Start the example DNS over QUIC server. Requires certificate and private key. The port can be overridden. ```console python examples/doq_server.py --certificate tests/ssl_cert.pem --private-key tests/ssl_key.pem --port 8053 ``` -------------------------------- ### Install aioquic and dependencies Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Install the aioquic library along with necessary dependencies for running examples. ```console pip install . dnslib jinja2 starlette wsproto ``` -------------------------------- ### Run HTTP/0.9 client example Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Perform a legacy HTTP/0.9 request using the example client. Requires CA certificates for verification. ```console python examples/http3_client.py --ca-certs tests/pycacert.pem --legacy-http https://localhost:4433/ ``` -------------------------------- ### Run HTTP/3 client example Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Perform an HTTP/3 request using the example client. Requires CA certificates for verification. ```console python examples/http3_client.py --ca-certs tests/pycacert.pem https://localhost:4433/ ``` -------------------------------- ### Install OpenSSL on Windows Source: https://github.com/aiortc/aioquic/blob/main/README.rst Install OpenSSL using the Chocolatey package manager. ```console choco install openssl ``` -------------------------------- ### Install dependencies on Debian/Ubuntu Source: https://github.com/aiortc/aioquic/blob/main/README.rst Install required development headers for OpenSSL and Python. ```console sudo apt install libssl-dev python3-dev ``` -------------------------------- ### Install OpenSSL on OS X Source: https://github.com/aiortc/aioquic/blob/main/README.rst Install OpenSSL using Homebrew. ```console brew install openssl ``` -------------------------------- ### Create a QUIC server with asyncio.serve Source: https://context7.com/aiortc/aioquic/llms.txt Starts a QUIC server using a callback-based stream handler. Requires a configured QuicConfiguration with certificate and key files. ```python import asyncio from aioquic.asyncio.server import serve from aioquic.quic.configuration import QuicConfiguration async def stream_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): """Handle incoming QUIC streams.""" # Read data from the stream data = await reader.read() print(f"Received: {data}") # Send response writer.write(b"Hello from server!") writer.write_eof() async def run_quic_server(): configuration = QuicConfiguration( is_client=False, alpn_protocols=["custom-protocol"], ) configuration.load_cert_chain("certificate.pem", "private-key.pem") server = await serve( host="::", # Listen on all interfaces (IPv4 and IPv6) port=4433, configuration=configuration, stream_handler=stream_handler, retry=True, # Enable RETRY packets for address validation ) print("QUIC server running on port 4433...") await asyncio.Future() # Run forever asyncio.run(run_quic_server()) ``` -------------------------------- ### Install aioquic via pip Source: https://github.com/aiortc/aioquic/blob/main/README.rst The standard command to install the library using the Python package manager. ```bash pip install aioquic ``` -------------------------------- ### Install dependencies on Alpine Linux Source: https://github.com/aiortc/aioquic/blob/main/README.rst Install required development headers for OpenSSL, Python, and compatibility libraries. ```console sudo apk add openssl-dev python3-dev bsd-compat-headers libffi-dev ``` -------------------------------- ### Create QUIC Client Connection with asyncio.connect Source: https://context7.com/aiortc/aioquic/llms.txt Establish a QUIC connection to a server using the `connect` async context manager. This handles DNS resolution, socket creation, and the TLS handshake. The example demonstrates creating streams, sending data, receiving responses, pinging, and closing the connection. ```python import asyncio from aioquic.asyncio.client import connect from aioquic.quic.configuration import QuicConfiguration async def quic_client_example(): configuration = QuicConfiguration(is_client=True) configuration.load_verify_locations("/etc/ssl/certs/ca-certificates.crt") async with connect( host="example.com", port=443, configuration=configuration, wait_connected=True, # Wait for TLS handshake to complete local_port=0, # Use any available local port ) as protocol: # Create a bidirectional stream reader, writer = await protocol.create_stream(is_unidirectional=False) # Send data writer.write(b"Hello, QUIC!") writer.write_eof() # Receive response response = await reader.read() print(f"Received: {response}") # Ping the server and wait for acknowledgment await protocol.ping() print("Ping acknowledged!") # Close the connection protocol.close() await protocol.wait_closed() asyncio.run(quic_client_example()) ``` -------------------------------- ### GET / Source: https://github.com/aiortc/aioquic/blob/main/examples/templates/index.html Returns the homepage of the aioquic server. ```APIDOC ## GET / ### Description Returns the homepage of the aioquic server. ### Method GET ### Endpoint / ### Response #### Success Response (200) - The content of the homepage. ``` -------------------------------- ### Open WebSocket over HTTP/3 Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Establish a WebSocket connection over HTTP/3 using the example client. Requires CA certificates for verification. ```console python examples/http3_client.py --ca-certs tests/pycacert.pem wss://localhost:4433/ws ``` -------------------------------- ### Set environment variables for OpenSSL on Windows Source: https://github.com/aiortc/aioquic/blob/main/README.rst Configure INCLUDE and LIB environment variables to point to the OpenSSL installation directory. ```console $Env:INCLUDE = "C:\Progra~1\OpenSSL\include" $Env:LIB = "C:\Progra~1\OpenSSL\lib" ``` -------------------------------- ### Implement an HTTP/3 Client with H3Connection Source: https://context7.com/aiortc/aioquic/llms.txt Extends QuicConnectionProtocol to handle HTTP/3 events and perform GET requests. Requires a configured QuicConfiguration with H3_ALPN. ```python import asyncio from aioquic.asyncio.client import connect from aioquic.asyncio.protocol import QuicConnectionProtocol from aioquic.h3.connection import H3Connection, H3_ALPN from aioquic.h3.events import HeadersReceived, DataReceived, H3Event from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import QuicEvent class Http3Client(QuicConnectionProtocol): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._http: H3Connection = None self._response_data = {} self._response_complete = {} def quic_event_received(self, event: QuicEvent) -> None: if self._http is None: self._http = H3Connection(self._quic) # Convert QUIC events to HTTP/3 events for h3_event in self._http.handle_event(event): self._h3_event_received(h3_event) def _h3_event_received(self, event: H3Event) -> None: if isinstance(event, HeadersReceived): stream_id = event.stream_id self._response_data[stream_id] = { "headers": dict(event.headers), "body": b"", } elif isinstance(event, DataReceived): stream_id = event.stream_id self._response_data[stream_id]["body"] += event.data if event.stream_ended: if stream_id in self._response_complete: self._response_complete[stream_id].set() async def get(self, url: str, authority: str, path: str) -> dict: """Perform an HTTP/3 GET request.""" stream_id = self._quic.get_next_available_stream_id() self._response_complete[stream_id] = asyncio.Event() # Send request headers self._http.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", authority.encode()), (b":path", path.encode()), (b"user-agent", b"aioquic-example"), ], end_stream=True, # No request body for GET ) self.transmit() # Wait for response await self._response_complete[stream_id].wait() return self._response_data[stream_id] async def http3_request(): config = QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN) config.load_verify_locations("/etc/ssl/certs/ca-certificates.crt") async with connect( "www.google.com", 443, configuration=config, create_protocol=Http3Client, ) as client: client = client # type: Http3Client response = await client.get( url="https://www.google.com/", authority="www.google.com", path="/", ) print(f"Status: {response['headers'].get(b':status')}") print(f"Body length: {len(response['body'])} bytes") asyncio.run(http3_request()) ``` -------------------------------- ### GET /NNNNN Source: https://github.com/aiortc/aioquic/blob/main/examples/templates/index.html Returns NNNNN bytes of plain text. Replace NNNNN with the desired number of bytes. ```APIDOC ## GET /NNNNN ### Description Returns NNNNN bytes of plain text. Replace NNNNN with the desired number of bytes. ### Method GET ### Endpoint /NNNNN ### Parameters #### Path Parameters - **NNNNN** (integer) - Required - The number of bytes to return. ``` -------------------------------- ### Launch Chromium/Chrome for demo server Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Launch Chromium or Chrome with specific flags to connect to the demo server, enabling experimental features and forcing QUIC. ```bash google-chrome \ --enable-experimental-web-platform-features \ --ignore-certificate-errors-spki-list=BSQJ0jkQ7wwhR7KvPZ+DSNk2XTZ/MS6xCbo9qu++VdQ= \ --origin-to-force-quic-on=localhost:4433 \ https://localhost:4433/ ``` -------------------------------- ### Handle Low-Level QUIC Connections Source: https://context7.com/aiortc/aioquic/llms.txt Demonstrates manual QUIC connection management without built-in I/O. Requires external handling of UDP datagram transmission and reception. ```python import time from aioquic.quic.connection import QuicConnection from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import ( HandshakeCompleted, StreamDataReceived, ConnectionTerminated ) # Create client connection (no network I/O - bring your own!) config = QuicConfiguration(is_client=True) config.server_name = "example.com" client = QuicConnection(configuration=config) # Initiate connection to server address server_addr = ("93.184.216.34", 443) now = time.time() client.connect(server_addr, now=now) # Get datagrams to send (these need to be sent over UDP) datagrams = client.datagrams_to_send(now=now) for data, addr in datagrams: print(f"Send {len(data)} bytes to {addr}") # Your I/O code: socket.sendto(data, addr) # When you receive data from the network: # client.receive_datagram(data, addr, now=time.time()) # Process events after receiving data while True: event = client.next_event() if event is None: break if isinstance(event, HandshakeCompleted): print(f"Connected! ALPN: {event.alpn_protocol}") # Send data on a stream stream_id = client.get_next_available_stream_id() client.send_stream_data(stream_id, b"Hello!", end_stream=True) elif isinstance(event, StreamDataReceived): print(f"Received on stream {event.stream_id}: {event.data}") elif isinstance(event, ConnectionTerminated): print(f"Connection closed: {event.reason_phrase}") ``` -------------------------------- ### CONNECT /ws Source: https://github.com/aiortc/aioquic/blob/main/examples/templates/index.html Establishes a WebSocket echo service. Requires the ':protocol' pseudo-header to be set to 'websocket'. ```APIDOC ## CONNECT /ws ### Description Runs a WebSocket echo service. You must set the _:protocol_ pseudo-header to _"websocket"_. ### Method CONNECT ### Endpoint /ws ### Request Headers - **_:protocol_** (string) - Required - Must be set to "websocket". ``` -------------------------------- ### CONNECT /wt Source: https://github.com/aiortc/aioquic/blob/main/examples/templates/index.html Establishes a WebTransport echo service. Requires the ':protocol' pseudo-header to be set to 'webtransport'. ```APIDOC ## CONNECT /wt ### Description Runs a WebTransport echo service. You must set the _:protocol_ pseudo-header to _"webtransport"_. ### Method CONNECT ### Endpoint /wt ### Request Headers - **_:protocol_** (string) - Required - Must be set to "webtransport". ``` -------------------------------- ### Enable QLOG Logging for QUIC Connections Source: https://context7.com/aiortc/aioquic/llms.txt Configures aioquic client to log connection events in QLOG format for debugging. Also shows how to log TLS secrets for Wireshark decryption. ```python import asyncio from aioquic.asyncio.client import connect from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.logger import QuicFileLogger async def connect_with_logging(): # Create QLOG logger - writes to specified directory quic_logger = QuicFileLogger("/tmp/qlog") config = QuicConfiguration( is_client=True, quic_logger=quic_logger, # Enable QLOG logging ) # Also log TLS secrets for Wireshark decryption secrets_log = open("/tmp/secrets.log", "a") config.secrets_log_file = secrets_log async with connect( "localhost", 4433, configuration=config, ) as client: reader, writer = await client.create_stream() writer.write(b"Test request") writer.write_eof() response = await reader.read() print(f"Response: {response}") secrets_log.close() # QLOG files are written to /tmp/qlog/*.qlog asyncio.run(connect_with_logging()) ``` -------------------------------- ### Run DNS over QUIC client Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Perform a DNS query using the DNS over QUIC client. Requires CA certificates and specifies query type, name, and port. ```console python examples/doq_client.py --ca-certs tests/pycacert.pem --query-type A --query-name quic.aiortc.org --port 8053 ``` -------------------------------- ### Implement WebSocket Client over HTTP/3 Source: https://context7.com/aiortc/aioquic/llms.txt Subclass QuicConnectionProtocol to handle HTTP/3 events and manage WebSocket state via wsproto. ```python import asyncio from aioquic.asyncio.client import connect from aioquic.asyncio.protocol import QuicConnectionProtocol from aioquic.h3.connection import H3Connection, H3_ALPN from aioquic.h3.events import HeadersReceived, DataReceived from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import QuicEvent import wsproto class WebSocketClient(QuicConnectionProtocol): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._http = None self._websocket = None self._ws_stream_id = None self._message_queue = asyncio.Queue() def quic_event_received(self, event: QuicEvent) -> None: if self._http is None: self._http = H3Connection(self._quic) for h3_event in self._http.handle_event(event): if isinstance(h3_event, HeadersReceived): # WebSocket handshake response received if h3_event.stream_id == self._ws_stream_id: self._websocket = wsproto.Connection( wsproto.ConnectionType.CLIENT ) elif isinstance(h3_event, DataReceived): if self._websocket and h3_event.stream_id == self._ws_stream_id: self._websocket.receive_data(h3_event.data) for ws_event in self._websocket.events(): if isinstance(ws_event, wsproto.events.TextMessage): self._message_queue.put_nowait(ws_event.data) async def open_websocket(self, path: str, authority: str): """Open WebSocket connection over HTTP/3.""" self._ws_stream_id = self._quic.get_next_available_stream_id() self._http.send_headers( stream_id=self._ws_stream_id, headers=[ (b":method", b"CONNECT"), (b":scheme", b"https"), (b":authority", authority.encode()), (b":path", path.encode()), (b":protocol", b"websocket"), (b"sec-websocket-version", b"13"), ], ) self.transmit() async def send_message(self, message: str): """Send WebSocket text message.""" data = self._websocket.send(wsproto.events.TextMessage(data=message)) self._http.send_data(self._ws_stream_id, data, end_stream=False) self.transmit() async def receive_message(self) -> str: """Receive WebSocket text message.""" return await self._message_queue.get() async def close_websocket(self): """Close WebSocket connection.""" data = self._websocket.send(wsproto.events.CloseConnection(code=1000)) self._http.send_data(self._ws_stream_id, data, end_stream=True) self.transmit() async def websocket_example(): config = QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN) async with connect( "localhost", 4433, configuration=config, create_protocol=WebSocketClient, ) as client: client = client # type: WebSocketClient await client.open_websocket("/chat", "localhost") await asyncio.sleep(0.5) # Wait for handshake await client.send_message("Hello WebSocket over HTTP/3!") response = await asyncio.wait_for(client.receive_message(), timeout=5.0) print(f"Received: {response}") await client.close_websocket() asyncio.run(websocket_example()) ``` -------------------------------- ### Implement an HTTP/3 Server with H3Connection Source: https://context7.com/aiortc/aioquic/llms.txt Uses H3Connection to process incoming HTTP/3 requests and send responses. Requires a QuicConfiguration with H3_ALPN and a valid certificate chain. ```python import asyncio from email.utils import formatdate import time from aioquic.asyncio.server import serve from aioquic.asyncio.protocol import QuicConnectionProtocol from aioquic.h3.connection import H3Connection, H3_ALPN from aioquic.h3.events import HeadersReceived, DataReceived from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import QuicEvent, ProtocolNegotiated class Http3Server(QuicConnectionProtocol): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._http = None self._request_body = {} def quic_event_received(self, event: QuicEvent) -> None: if isinstance(event, ProtocolNegotiated): self._http = H3Connection(self._quic) if self._http is not None: for h3_event in self._http.handle_event(event): self._handle_h3_event(h3_event) def _handle_h3_event(self, event): if isinstance(event, HeadersReceived): stream_id = event.stream_id headers = dict(event.headers) method = headers.get(b":method", b"GET").decode() path = headers.get(b":path", b"/").decode() print(f"Request: {method} {path}") # Accumulate request body self._request_body[stream_id] = b"" if event.stream_ended: self._send_response(stream_id, path) elif isinstance(event, DataReceived): stream_id = event.stream_id self._request_body[stream_id] += event.data if event.stream_ended: self._send_response(stream_id, "/") def _send_response(self, stream_id: int, path: str): body = f"

Hello from HTTP/3!

Path: {path}

" body_bytes = body.encode() # Send response headers self._http.send_headers( stream_id=stream_id, headers=[ (b":status", b"200"), (b"content-type", b"text/html"), (b"content-length", str(len(body_bytes)).encode()), (b"server", b"aioquic"), (b"date", formatdate(time.time(), usegmt=True).encode()), ], ) # Send response body self._http.send_data( stream_id=stream_id, data=body_bytes, end_stream=True, ) self.transmit() async def run_server(): config = QuicConfiguration( is_client=False, alpn_protocols=H3_ALPN, ) config.load_cert_chain("cert.pem", "key.pem") await serve( host="::", port=4433, configuration=config, create_protocol=Http3Server, ) await asyncio.Future() asyncio.run(run_server()) ``` -------------------------------- ### Configure QUIC Connections with QuicConfiguration Source: https://context7.com/aiortc/aioquic/llms.txt Configure QUIC connections for client or server roles, including TLS settings, flow control, and protocol versions. Load CA certificates for verification or disable verification for testing. Server configurations can load TLS certificates and keys. ```python from aioquic.quic.configuration import QuicConfiguration from aioquic.h3.connection import H3_ALPN import ssl # Client configuration client_config = QuicConfiguration( is_client=True, alpn_protocols=H3_ALPN, # ["h3"] for HTTP/3 max_data=1048576, # Connection-wide flow control limit (1MB) max_stream_data=1048576, # Per-stream flow control limit (1MB) idle_timeout=60.0, # Idle timeout in seconds congestion_control_algorithm="reno", # or "cubic" ) # Load CA certificates for server verification client_config.load_verify_locations(cafile="/path/to/ca-certificates.crt") # Disable certificate verification (for testing only) client_config.verify_mode = ssl.CERT_NONE # Server configuration with TLS certificate server_config = QuicConfiguration( is_client=False, alpn_protocols=H3_ALPN, max_datagram_frame_size=65536, # Enable DATAGRAM frames ) server_config.load_cert_chain( certfile="/path/to/certificate.pem", keyfile="/path/to/private-key.pem" ) # Enable secrets logging for Wireshark debugging with open("secrets.log", "a") as f: client_config.secrets_log_file = f ``` -------------------------------- ### QUIC Configuration API Source: https://github.com/aiortc/aioquic/blob/main/docs/quic.rst Information regarding the QuicConfiguration class for setting up QUIC connections. ```APIDOC ## QuicConfiguration API ### Description Defines the configuration options for QUIC connections. ### Class `aioquic.quic.configuration.QuicConfiguration` ### Members (Details of members are not provided in the source text, but the `automodule` directive indicates they are available for inspection.) ``` -------------------------------- ### aioquic Asyncio Server API Source: https://github.com/aiortc/aioquic/blob/main/docs/asyncio.rst Provides functions for creating and managing QUIC servers using asyncio. ```APIDOC ## serve ### Description Starts a QUIC server that listens for incoming connections. ### Method `serve` ### Endpoint N/A (Function call) ### Parameters Refer to the `aioquic.asyncio.serve` function documentation for detailed parameter information. ``` -------------------------------- ### Connect with 0-RTT Session Resumption Source: https://context7.com/aiortc/aioquic/llms.txt Shows how to load a saved session ticket for faster connection establishment using 0-RTT. The `session_ticket_handler` saves new tickets. Set `wait_connected=False` to enable early data transmission. ```python import asyncio import pickle from aioquic.asyncio.client import connect from aioquic.quic.configuration import QuicConfiguration from aioquic.tls import SessionTicket # Storage for session ticket session_ticket_file = "session_ticket.pkl" def save_session_ticket(ticket: SessionTicket) -> None: """Callback invoked when a new session ticket is received.""" with open(session_ticket_file, "wb") as f: pickle.dump(ticket, f) print("Session ticket saved for future 0-RTT") async def connect_with_0rtt(): config = QuicConfiguration(is_client=True) # Load previous session ticket if available try: with open(session_ticket_file, "rb") as f: config.session_ticket = pickle.load(f) print("Loaded session ticket for 0-RTT") except FileNotFoundError: print("No session ticket, doing full handshake") async with connect( "localhost", 4433, configuration=config, session_ticket_handler=save_session_ticket, wait_connected=False, # Don't wait - enable 0-RTT! ) as client: # Can start sending data immediately if 0-RTT is available stream_id = client._quic.get_next_available_stream_id() client._quic.send_stream_data(stream_id, b"Early data!", end_stream=True) client.transmit() # Wait for handshake to verify if 0-RTT was accepted await client.wait_connected() print("Handshake complete") asyncio.run(connect_with_0rtt()) ``` -------------------------------- ### aioquic Asyncio Client API Source: https://github.com/aiortc/aioquic/blob/main/docs/asyncio.rst Provides functions for establishing and managing QUIC client connections using asyncio. ```APIDOC ## connect ### Description Establishes a QUIC connection to a remote host. ### Method `connect` ### Endpoint N/A (Function call) ### Parameters Refer to the `aioquic.asyncio.connect` function documentation for detailed parameter information. ``` -------------------------------- ### Set environment variables for OpenSSL on OS X Source: https://github.com/aiortc/aioquic/blob/main/README.rst Configure CFLAGS and LDFLAGS to link against the Homebrew-installed OpenSSL. ```console export CFLAGS=-I$(brew --prefix openssl)/include export LDFLAGS=-L$(brew --prefix openssl)/lib ``` -------------------------------- ### POST /echo Source: https://github.com/aiortc/aioquic/blob/main/examples/templates/index.html Echoes the request data back to the client. ```APIDOC ## POST /echo ### Description Returns the request data sent by the client. ### Method POST ### Endpoint /echo ### Request Body - **data** (any) - Required - The data to be echoed back. ### Request Example { "message": "Hello, aioquic!" } ### Response #### Success Response (200) - **echoed_data** (any) - The data received in the request body. ``` -------------------------------- ### QUIC Events API Source: https://github.com/aiortc/aioquic/blob/main/docs/quic.rst Documentation for various QUIC events that can be received or emitted. ```APIDOC ## QUIC Events API ### Description Defines the different types of events that can occur within a QUIC connection. ### Base Class `aioquic.quic.events.QuicEvent` ### Event Types - `ConnectionTerminated` - `HandshakeCompleted` - `PingAcknowledged` - `StopSendingReceived` - `StreamDataReceived` - `StreamReset` ### Members (Details of members for each event type are not provided in the source text, but the `automodule` directive indicates they are available for inspection.) ``` -------------------------------- ### H3Event Classes Source: https://github.com/aiortc/aioquic/blob/main/docs/h3.rst Documentation for various HTTP/3 event classes. ```APIDOC ## H3Event Classes ### Description Represents different types of events that can occur during HTTP/3 communication. ### Base Class H3Event ### Event Types - DatagramReceived - DataReceived - HeadersReceived - PushPromiseReceived - WebTransportStreamDataReceived ### Members (Members for each event class are detailed in the source documentation) ``` -------------------------------- ### aioquic Common QUIC Protocol Source: https://github.com/aiortc/aioquic/blob/main/docs/asyncio.rst Details the common QUIC connection protocol class used by both clients and servers. ```APIDOC ## QuicConnectionProtocol ### Description Represents a common QUIC connection protocol. ### Members Refer to the `aioquic.asyncio.QuicConnectionProtocol` class documentation for a list of its members and their functionalities. ``` -------------------------------- ### QUIC Connection API Source: https://github.com/aiortc/aioquic/blob/main/docs/quic.rst Details about the QuicConnection class, which manages QUIC connections. ```APIDOC ## QuicConnection API ### Description Provides methods for managing QUIC connections. The API user is responsible for handling I/O operations. ### Class `aioquic.quic.connection.QuicConnection` ### Members (Details of members are not provided in the source text, but the `automodule` directive indicates they are available for inspection.) ``` -------------------------------- ### Send and Receive Datagram Frames Source: https://context7.com/aiortc/aioquic/llms.txt Demonstrates sending an unreliable datagram frame and waiting for a response. Ensure DATAGRAM support is enabled in the QuicConfiguration. ```python import asyncio from aioquic.asyncio.client import connect from aioquic.asyncio.protocol import QuicConnectionProtocol from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import QuicEvent, DatagramFrameReceived class DatagramClient(QuicConnectionProtocol): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.datagram_received = asyncio.Event() self.last_datagram = None def quic_event_received(self, event: QuicEvent) -> None: if isinstance(event, DatagramFrameReceived): self.last_datagram = event.data self.datagram_received.set() print(f"Received datagram: {event.data}") async def datagram_example(): config = QuicConfiguration( is_client=True, max_datagram_frame_size=65536, # Enable DATAGRAM support ) async with connect( "localhost", 4433, configuration=config, create_protocol=DatagramClient, ) as client: client = client # type: DatagramClient # Send unreliable datagram client._quic.send_datagram_frame(b"ping") client.transmit() # Wait for response datagram await asyncio.wait_for(client.datagram_received.wait(), timeout=5.0) print(f"Got response: {client.last_datagram}") asyncio.run(datagram_example()) ``` -------------------------------- ### Manage QUIC connection state with QuicConnectionProtocol Source: https://context7.com/aiortc/aioquic/llms.txt Implements custom protocol logic by subclassing QuicConnectionProtocol and overriding quic_event_received. Used to handle handshake, stream data, and datagram events. ```python import asyncio from aioquic.asyncio.protocol import QuicConnectionProtocol from aioquic.quic.connection import QuicConnection from aioquic.quic.events import ( QuicEvent, HandshakeCompleted, StreamDataReceived, ConnectionTerminated, DatagramFrameReceived ) class CustomProtocol(QuicConnectionProtocol): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.connected = asyncio.Event() def quic_event_received(self, event: QuicEvent) -> None: """Handle QUIC-level events.""" if isinstance(event, HandshakeCompleted): print(f"Connected! ALPN: {event.alpn_protocol}") print(f"Session resumed: {event.session_resumed}") print(f"0-RTT accepted: {event.early_data_accepted}") self.connected.set() elif isinstance(event, StreamDataReceived): print(f"Stream {event.stream_id}: {event.data}") if event.end_stream: print(f"Stream {event.stream_id} ended") elif isinstance(event, DatagramFrameReceived): print(f"Datagram received: {event.data}") # Echo back the datagram self._quic.send_datagram_frame(event.data) self.transmit() elif isinstance(event, ConnectionTerminated): print(f"Connection closed: code={event.error_code}, " f"reason={event.reason_phrase}") # Using the custom protocol with connect() from aioquic.asyncio.client import connect from aioquic.quic.configuration import QuicConfiguration async def main(): config = QuicConfiguration(is_client=True) config.verify_mode = None # Skip verification for demo async with connect( "localhost", 4433, configuration=config, create_protocol=CustomProtocol, # Use custom protocol class ) as protocol: protocol = protocol # type: CustomProtocol await protocol.connected.wait() # Send data on a stream stream_id = protocol._quic.get_next_available_stream_id() protocol._quic.send_stream_data(stream_id, b"Hello!", end_stream=True) protocol.transmit() await asyncio.sleep(1) asyncio.run(main()) ``` -------------------------------- ### H3Error Classes Source: https://github.com/aiortc/aioquic/blob/main/docs/h3.rst Documentation for exceptions raised by the HTTP/3 API. ```APIDOC ## H3Error Classes ### Description Custom exceptions used within the HTTP/3 API to indicate errors. ### Base Exception H3Error ### Specific Exceptions - InvalidStreamTypeError - NoAvailablePushIDError ### Members (Members for each exception class are detailed in the source documentation) ``` -------------------------------- ### QUIC Logger API Source: https://github.com/aiortc/aioquic/blob/main/docs/quic.rst Details on the QuicLogger class for logging QUIC events. ```APIDOC ## QuicLogger API ### Description Provides functionality for logging QUIC events during connection lifetime. ### Class `aioquic.quic.logger.QuicLogger` ### Members (Details of members are not provided in the source text, but the `automodule` directive indicates they are available for inspection.) ``` -------------------------------- ### Generate certificate fingerprint Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst Command to generate the SHA256 base64 encoded public key fingerprint for a given certificate, used with `--ignore-certificate-errors-spki-list`. ```bash openssl x509 -in tests/ssl_cert.pem -pubkey -noout | \ openssl pkey -pubin -outform der | \ openssl dgst -sha256 -binary | \ openssl enc -base64 ``` -------------------------------- ### WebTransport echo service client Source: https://github.com/aiortc/aioquic/blob/main/examples/README.rst JavaScript code to connect to a WebTransport echo service, send data, and receive a response. Requires an active WebTransport connection. ```javascript let transport = new WebTransport('https://localhost:4433/wt'); await transport.ready; let stream = await transport.createBidirectionalStream(); let reader = stream.readable.getReader(); let writer = stream.writable.getWriter(); await writer.write(new Uint8Array([65, 66, 67])); let received = await reader.read(); await transport.close(); console.log('received', received); ``` -------------------------------- ### H3Connection API Source: https://github.com/aiortc/aioquic/blob/main/docs/h3.rst Details regarding the H3Connection class for managing HTTP/3 connections. ```APIDOC ## H3Connection API ### Description Provides methods for managing HTTP/3 connections. ### Class H3Connection ### Methods (Members of H3Connection are detailed in the source documentation) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.