SSE Connection Monitor

Pattern for monitoring Server-Sent Events (SSE) connections with visual status indicators and automatic reconnection

Connection Status Configuration

The SSEConnectionConfig dataclass defines configuration options for the SSE connection monitor.

---

source

SSEConnectionConfig

def SSEConnectionConfig(
    max_reconnect_attempts:int=10, reconnect_delay:int=1000, max_backoff_multiplier:int=5,
    monitor_visibility:bool=True, log_to_console:bool=True
)->None:

Configuration for SSE connection monitoring.

Status Indicator Creation

Helper function to create status indicators for different connection states.

---

source

create_connection_status_indicators

def create_connection_status_indicators(
    status_size:str='sm', # Size of status indicator dot (xs, sm, md, lg)
    show_text:bool=True, # Whether to show status text
    text_size:str='text-sm', # Text size class
    hide_text_on_mobile:bool=True, # Hide text on small screens
)->Dict: # Dictionary of status state to indicator element

Create status indicator elements for different connection states.

SSE Connection Monitor Script

The SSEConnectionMonitorScript function creates a JavaScript script that monitors SSE connection status.

Key features: - Monitors HTMX SSE events (open, error, close, message) - Updates visual status indicators - Automatic reconnection with exponential backoff - Handles server shutdown gracefully - Tab visibility awareness (reconnects when tab becomes visible) - Detects OOB swaps that remove SSE element

---

source

SSEConnectionMonitorScript

def SSEConnectionMonitorScript(
    connection_id:str, # Unique identifier for this SSE connection
    status_indicators:Dict, # Status indicator elements for each state
    config:Optional=None, # Configuration options
)->FT: # Script element with monitoring code

Create a script that monitors SSE connection status and manages reconnection.

Complete SSE Connection Monitor

The SSEConnectionMonitor function creates a complete connection monitoring system with status indicator and monitoring script.

---

source

SSEConnectionMonitor

def SSEConnectionMonitor(
    connection_id:str, # Unique identifier for this SSE connection
    status_size:str='sm', # Size of status indicator
    show_text:bool=True, # Whether to show status text
    hide_text_on_mobile:bool=True, # Hide text on small screens
    config:Optional=None, # Configuration options
    container_cls:Optional=None, # Additional CSS classes for status container
)->tuple: # Tuple of (status_container, monitor_script)

Create a complete SSE connection monitoring system.

Usage Examples

Here are complete examples showing different SSE monitoring use cases:

# Example 1: Simple connection monitor
status_container, monitor_script = SSEConnectionMonitor(
    connection_id="simple-stream"
)

print("Example 1: Simple connection monitor")
print(f"Status container ID: {InteractionHtmlIds.sse_status('simple-stream')}")
print(f"SSE element ID: {InteractionHtmlIds.sse_element('simple-stream')}")

Example 1: Simple connection monitor
Status container ID: sse-status-simple-stream
SSE element ID: sse-element-simple-stream

# Example 2: Custom configuration
config = SSEConnectionConfig(
    max_reconnect_attempts=5,
    reconnect_delay=2000,
    max_backoff_multiplier=3,
    log_to_console=False  # Disable logging in production
)

status_container, monitor_script = SSEConnectionMonitor(
    connection_id="dashboard",
    status_size="md",
    config=config
)

print("Example 2: Custom configuration")
print(f"Max reconnect attempts: {config.max_reconnect_attempts}")
print(f"Reconnect delay: {config.reconnect_delay}ms")

Example 2: Custom configuration
Max reconnect attempts: 5
Reconnect delay: 2000ms

# Example 3: Custom status indicators
from cjm_fasthtml_daisyui.components.data_display.badge import badge, badge_colors

# Create custom indicators using badges instead of status dots
custom_indicators = {
    'active': Span("Connected", cls=str(combine_classes(badge, badge_colors.success))),
    'disconnected': Span("Offline", cls=str(combine_classes(badge, badge_colors.warning))),
    'error': Span("Error", cls=str(combine_classes(badge, badge_colors.error))),
    'reconnecting': Span("Connecting", cls=str(combine_classes(badge, badge_colors.info))),
}

# Create monitor script with custom indicators
monitor_script = SSEConnectionMonitorScript(
    connection_id="custom",
    status_indicators=custom_indicators
)

status_container = Div(
    custom_indicators['reconnecting'],
    id=InteractionHtmlIds.sse_status("custom")
)

print("Example 3: Custom status indicators using badges")

Example 3: Custom status indicators using badges

Complete Page Example

Here’s a complete example showing how to use the SSE connection monitor in a FastHTML application:

from fasthtml.common import Div, H1, P
from cjm_fasthtml_daisyui.components.data_display.card import card, card_body
from cjm_fasthtml_daisyui.utilities.semantic_colors import bg_dui
from cjm_fasthtml_tailwind.utilities.flexbox_and_grid import flex_display, items, justify, gap
from cjm_fasthtml_tailwind.utilities.spacing import p as padding

def create_dashboard_page():
    """Example dashboard page with SSE monitoring."""
    connection_id = "dashboard"
    
    # Create connection monitor
    status_container, monitor_script = SSEConnectionMonitor(
        connection_id=connection_id,
        status_size="sm"
    )
    
    return Div(
        # Header with connection status
        Div(
            Div(
                H1("Dashboard"),
                status_container,  # Connection status indicator
                cls=combine_classes(
                    flex_display,
                    items.center,
                    justify.between,
                    padding.b(4)
                )
            )
        ),
        
        # SSE connection element (receives updates)
        Div(
            # Content will be updated via SSE
            P("Waiting for updates..."),
            hx_ext="sse",
            sse_connect="/stream/dashboard",
            sse_swap="update",
            id=InteractionHtmlIds.sse_element(connection_id),
            cls=str(combine_classes(card, card_body, bg_dui.base_100))
        ),
        
        # Monitor script (must be included)
        monitor_script
    )

# Example route handler
print("Example: Complete dashboard page with SSE monitoring")
print("Route: /dashboard")
print("SSE Endpoint: /stream/dashboard")

Example: Complete dashboard page with SSE monitoring
Route: /dashboard
SSE Endpoint: /stream/dashboard

Server-Side SSE Implementation

Here’s how to implement the server-side SSE endpoint:

from fasthtml.common import *
from starlette.responses import StreamingResponse
import asyncio

@app.get("/stream/dashboard")
async def stream_dashboard():
    """SSE endpoint for dashboard updates."""
    
    async def event_generator():
        try:
            while True:
                # Generate your data
                data = get_dashboard_data()
                
                # Send SSE event
                yield f"event: update\n"
                yield f"data: {data}\n\n"
                
                # Wait before next update
                await asyncio.sleep(1)
                
        except asyncio.CancelledError:
            # Send close message before shutting down
            yield f"event: close\n"
            yield f"data: Server shutting down\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

Graceful Shutdown

To signal shutdown to clients, send a ‘close’ event:

# In your shutdown handler
async def shutdown_sse_connections():
    """Send close event to all SSE clients."""
    for client in active_connections:
        await client.send("event: close\n")
        await client.send("data: Server shutting down\n\n")

OOB Swap for Shutdown

Alternatively, use OOB swap to remove the SSE element:

# Send before shutdown
Div(
    id=InteractionHtmlIds.sse_element(connection_id),
    hx_swap_oob="outerHTML"
)