INNER CODE UNIT · Python

ControlRequest

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

class ControlRequest:
    identifier: int | None
    scope: RequestScope
    business: asyncio.Event
    task: asyncio.Task[None] | None = None
    thread: Future[None] | None = None


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


class ControlOutbox:
    def __init__(self) -> None:
        self.loop = asyncio.get_running_loop()

View source record →