INNER CODE UNIT · Python

_wake_locked

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

    def _wake_locked(self) -> None:
        if not self.wake_pending:
            self.wake_pending = True
            self.loop.call_soon_threadsafe(self._wake)

    def _wake(self) -> None:
        with self.lock:
            self.wake_pending = False
            closed = self.closed
        self.available.set()
        if closed:
            self.closing.set()

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

View source record →