aioquic で echo サーバー

01-README.md

aioquic で echo サーバー

uv init
uv add aioquic

02-server.py

import asyncio
import ssl
from aioquic.asyncio import serve, QuicConnectionProtocol
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import StreamDataReceived


class EchoServerProtocol(QuicConnectionProtocol):

    def quic_event_received(self, event):
        if isinstance(event, StreamDataReceived):
            # 受信データをそのまま送り返す
            self._quic.send_stream_data(
                event.stream_id,
                event.data,
                end_stream=event.end_stream,
            )
            self.transmit()


async def main():
    configuration = QuicConfiguration(
        is_client=False,
        alpn_protocols=["hq-interop"],  # HTTP/3ではない
    )

    configuration.load_cert_chain("cert.pem", "key.pem")

    await serve(
        host="::",
        port=4433,
        configuration=configuration,
        create_protocol=EchoServerProtocol,
    )

    print("QUIC Echo Server started on udp/4433")
    await asyncio.Future()  # run forever


if __name__ == "__main__":
    asyncio.run(main())

03-client.py

import asyncio
import ssl
from aioquic.asyncio import connect, QuicConnectionProtocol
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import StreamDataReceived


class EchoClientProtocol(QuicConnectionProtocol):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._done = asyncio.get_running_loop().create_future()

    def quic_event_received(self, event):
        if isinstance(event, StreamDataReceived):
            print("echo:", event.data.decode(errors="replace"), end="")
            if event.end_stream and not self._done.done():
                self._done.set_result(True)

    async def wait_echo(self, timeout: float = 3.0):
        await asyncio.wait_for(self._done, timeout=timeout)


async def run():
    cfg = QuicConfiguration(
        is_client=True,
        alpn_protocols=["hq-interop"],
    )
    cfg.verify_mode = ssl.CERT_NONE

    async with connect(
        "::",
        4433,
        configuration=cfg,
        create_protocol=EchoClientProtocol,
        wait_connected=True,
    ) as proto:
        stream_id = proto._quic.get_next_available_stream_id()
        proto._quic.send_stream_data(stream_id, b"hello over quic\n", end_stream=True)
        proto.transmit()

        await proto.wait_echo(timeout=3.0)

        proto._quic.close()
        await asyncio.sleep(0.05)


async def main():
    # ★ connect 自体が固まるケースも潰す
    await asyncio.wait_for(run(), timeout=5.0)


if __name__ == "__main__":
    asyncio.run(main())