Function SSEMessage

robyn/responses.py:195–234

Optimized SSE message formatting with minimal allocations. :param data: The message data :param event: Optional event type :param id: Optional event ID :param retry: Optional retry time in milliseconds :return: SSE-formatted string

(data: str, event: str | None = None, id: str | None = None, retry: int | None = None)

def SSEMessage(data: str, event: str | None = None, id: str | None = None, retry: int | None = None) -> str:
    """
    Optimized SSE message formatting with minimal allocations.

    :param data: The message data
    :param event: Optional event type
    :param id: Optional event ID
    :param retry: Optional retry time in milliseconds
    :return: SSE-formatted string
    """
    # Pre-calculate size to avoid multiple string concatenations
    parts = []

    # Add optional fields first
    if event:
        parts.append(f"event: {event}\n")
    if id:
        parts.append(f"id: {id}\n")
    if retry:
        parts.append(f"retry: {retry}\n")

    # Handle data with optimized multi-line processing
    if data:
        data_str = str(data)
        # Fast path for single-line data (most common case)
        if "\n" not in data_str and "\r" not in data_str:
            parts.append(f"data: {data_str}\n")
        else:
            # Multi-line data handling
            normalized_data = data_str.replace("\r\n", "\n").replace("\r", "\n")
            for line in normalized_data.split("\n"):
                parts.append(f"data: {line}\n")
    else:
        parts.append("data: \n")

    # Add the required double newline terminator
    parts.append("\n")

    # Single join operation for optimal performance
    return "".join(parts)