INNER CODE UNIT · Python

fail

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

    def fail(self) -> None:
        with self.lock:
            self.closed = True
            self._wake_locked()

    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)

View source record →