INNER CODE UNIT · Python

ControlOutbox

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

class ControlOutbox:
    def __init__(self) -> None:
        self.loop = asyncio.get_running_loop()
        self.lock = threading.Condition()
        self.encoder = threading.Lock()
        self.items: deque[OutgoingFrame] = deque()
        self.responses: set[asyncio.Future[None]] = set()
        self.available = asyncio.Event()
        self.closing = asyncio.Event()
        self.closed = False
        self.wake_pending = False
        self.frames = 0
        self.used = 0
        self.producers = 0

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

View source record →