Deep Dive into Server-Sent Events (SSE)

What is SSE ?

Server-Sent Events (SSE) is a mechanism that allows the server to push real-time updates to the client over a single HTTP connection. Unlike WebSockets, which support bidirectional communication, SSE is unidirectional—the server sends data, and the client listens.

---

1. How SSE Works

Step-by-Step Flow

1. The client (browser or frontend app) creates an EventSource connection to an SSE endpoint.

2. The server keeps the connection open and streams data as events.

3. The client automatically handles reconnections if the connection is lost.

4. The server sends data in a special format where each event starts with data: and ends with \n\n.

Example SSE Data Format

data: {"message": "Hello World"}

The browser treats this as an event and triggers onmessage

---

2. SSE vs. WebSockets vs. Long Polling

Feature	SSE	WebSockets	Long Polling
Connection	One-way (server → client)	Two-way (server ↔ client)	Repeated requests
Performance	Efficient (persistent connection)	Best for interactive apps	Inefficient (many requests)
Reconnection	Automatic	Manual handling needed	New request needed
Use Cases	Notifications, updates, logs, AI streaming	Chat, gaming, collaboration apps	Legacy real-time updates

---

3. When to Use SSE?

SSE is ideal for: ✅ Live notifications (e.g., stock updates, sports scores)
✅ Real-time analytics dashboards
✅ Live AI model responses (e.g., streaming chatbot messages)
✅ IoT device monitoring
✅ Event-driven architectures (e.g., order tracking updates)

Not ideal for:
❌ Bidirectional communication (Use WebSockets)
❌ Very high-frequency updates (e.g., 1000s/sec)

---

4. Advanced SSE Features

A. Custom Events in SSE

Instead of only using onmessage, SSE supports named events:

Server (FastAPI)

@app.get("/custom_events")
async def custom_events():
    async def event_stream():
        for i in range(5):
            await asyncio.sleep(2)
            yield f"event: update\ndata: {json.dumps({'message': f'Update {i+1}'})}\n\n"
        yield "event: done\ndata: {}\n\n"  # Custom event when completed
    return StreamingResponse(event_stream(), media_type="text/event-stream")

Client (JavaScript)

const eventSource = new EventSource("http://127.0.0.1:8000/custom_events");

eventSource.addEventListener("update", (event) => {
    console.log("Update Event:", JSON.parse(event.data));
});

eventSource.addEventListener("done", () => {
    console.log("Streaming completed.");
    eventSource.close();
});

---

B. Filtering SSE Data for Specific Users

Server

@app.get("/user_stream")
async def user_stream(user_id: str):
    async def event_stream():
        for i in range(5):
            await asyncio.sleep(2)
            yield f"data: {json.dumps({'user_id': user_id, 'message': f'Update {i+1}'})}\n\n"
    return StreamingResponse(event_stream(), media_type="text/event-stream")

Client

const eventSource = new EventSource("http://127.0.0.1:8000/user_stream?user_id=123");
eventSource.onmessage = (event) => {
    console.log("Received:", JSON.parse(event.data));
};

✅ Now, only events meant for user_id=123 are received.

---

C. Securing SSE with Authentication

SSE doesn’t support headers for authentication in EventSource. Instead, use:

1. URL token authentication
2. Cookie-based authentication
3. JWT Authentication with FastAPI Dependencies

Example: JWT Authentication

from fastapi import Depends

def verify_token(token: str = Depends(authenticate_user)):  # Your JWT auth function
    return token

@app.get("/secure_events")
async def secure_events(token: str = Depends(verify_token)):
    return StreamingResponse(event_stream(), media_type="text/event-stream")

✅ Now, only authenticated users can access the SSE stream.

---

D. Optimizing SSE Performance

1. Use gzip compression: Reduces payload size.
2. Limit open connections: Too many SSE connections may overload the server.
3. Use Redis Pub/Sub: For handling multiple clients with event broadcasting.

Using Redis for Broadcasting Events

import aioredis

redis = await aioredis.from_url("redis://localhost")

@app.get("/events")
async def sse_endpoint():
    async def event_stream():
        async with redis.pubsub() as pubsub:
            await pubsub.subscribe("updates")
            while True:
                message = await pubsub.get_message(ignore_subscribe_messages=True)
                if message:
                    yield f"data: {message['data'].decode()}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

✅ Now, multiple clients receive the same live updates.

---

5. Conclusion

🚀 SSE is a lightweight, efficient solution for real-time updates when only server-to-client communication is needed.
💡 Use SSE for live dashboards, notifications, streaming responses, and AI model outputs.
🔧 For bidirectional communication, consider WebSockets.