INNER CODE UNIT · Python

is_active

ok-oldking/ok-script · ok/ui/web/app.py:99

    def is_active(self, session_key, token):
        with self._lock:
            current = self._active.get(session_key)
            return current is not None and current[0] is token

    def unregister(self, session_key, token):
        with self._lock:
            current = self._active.get(session_key)
            if current is not None and current[0] is token:
                self._active.pop(session_key, None)


async def _send_websocket_payload(websocket, payload):
    """Send an event, treating Starlette's post-close race as disconnect."""
    try:
        await websocket.send_json(payload)
    except RuntimeError as error:
        message = str(error)

View source record →