Function _process_ws_message

dash/backends/ws.py:256–299

Process a single WebSocket message from the queue. Args: q: The async queue to read from send_text: Async function to send text data over WebSocket messages: List to accumulate messages for batching (mutated in place) batch_delay: Batch delay in seconds Retu

(
    q: "janus._AsyncQueueProxy[str]",
    send_text: Callable[[str], Any],
    messages: list[str],
    batch_delay: float,
)

async def _process_ws_message(
    q: "janus._AsyncQueueProxy[str]",
    send_text: Callable[[str], Any],
    messages: list[str],
    batch_delay: float,
) -> bool:
    """Process a single WebSocket message from the queue.

    Args:
        q: The async queue to read from
        send_text: Async function to send text data over WebSocket
        messages: List to accumulate messages for batching (mutated in place)
        batch_delay: Batch delay in seconds

    Returns:
        True to continue processing, False to stop the sender loop.
    """
    timeout = batch_delay if messages else None
    try:
        msg = await asyncio.wait_for(q.get(), timeout=timeout)
    except asyncio.TimeoutError:
        success = await _send_batched(send_text, messages)
        messages.clear()
        return success

    if msg == SHUTDOWN_SIGNAL:
        if messages:
            await _send_batched(send_text, messages)
        return False

    if msg == FLUSH_SIGNAL:
        success = not messages or await _send_batched(send_text, messages)
        messages.clear()
        return success

    if not batch_delay:
        try:
            await send_text(msg)
        except Exception:  # pylint: disable=broad-exception-caught
            return False  # WebSocketDisconnect, RuntimeError, etc.
    else:
        messages.append(msg)

    return True


async def _send_batched(send_text: Callable[[str], Any], messages: list) -> bool: