Class ASGIRunner

benchmarks/_utils.py:44–53

class ASGIRunner:
    def __init__(self) -> None:
        self.loop = asyncio.new_event_loop()

    def run(self, app: ASGIApp, scope: Scope, receive: Receive = unreadable_receive) -> list[Message]:
        return self.loop.run_until_complete(run_asgi(app, scope, receive))

    def close(self) -> None:
        self.loop.run_until_complete(self.loop.shutdown_asyncgens())
        self.loop.close()