INNER CODE UNIT · Python

OutgoingFrame

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

class OutgoingFrame:
    text: str
    cost: int
    completion: asyncio.Future[None] | None


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

View source record →