INNER CODE UNIT · Python

register_response

ImFeH2/huddol · core/src/fora/adapters/websocket/server.py:106

    def register_response(self) -> asyncio.Future[None]:
        result = self.loop.create_future()
        self.responses.add(result)
        return result

    def complete(self, result: asyncio.Future[None]) -> None:
        if not result.done():
            result.set_result(None)
        self.responses.discard(result)

    def broadcast(self, payload: dict[str, Any]) -> None:
        self.put(payload)

    def put(
        self,
        payload: dict[str, Any],
        completion: asyncio.Future[None] | None = None,
    ) -> None:

View source record →