BaseStreamingService

def __init__(self, name: str) -> None:
    self.mode = name

@abstractmethod
async def start(self) -> None:
    """Set up resources and run the service's loops until stopped."""

@abstractmethod
async def stop(self) -> None:
    """Clean up resources and stop the service's loops."""

@abstractmethod
def register_routes(self, api_prefix: str, main_router: web.UrlDispatcher) -> None:
    """Register the service's absolute paths directly on the main router.

    Args:
        api_prefix: Deployment subfolder prefix (empty when serving at root).
        main_router: The application's router to register routes on.
    """
    pass

@abstractmethod
async def announce_print_document(self, name: str, size: int) -> None:
    """Tell the controller pages that `name` waits in the print spool."""

@abstractmethod
async def sessions(self) -> List[Dict[str, Any]]:
    """The pages connected to this transport: `id`, `transport`, `role`,
    `slot`, `display`, `connected_at` and `rtt_ms`, the same keys on both."""

@abstractmethod
async def disconnect_session(self, session_id: str) -> bool:
    """Close the page `id` names; False when it is not connected here."""

def uplink_session_conns(self) -> List[Tuple[Any, Optional[str], Optional[str]]]:
    """``(websocket, session token, peer ip)`` per connected client
    session, for upload uplink gauging (`UplinkGauge`).

    The websocket must send ``ping()`` frames, and the service's message
    loop must run with autoping off, answer PING itself, and hand every
    PONG to `note_pong` — the gauge's clock is dead without that. Both
    transports implement this; an upload gauged on one side but not the
    other is a parity bug. Default: nothing to gauge.
    """
    return []