Server-Sent Events (SSE)

After learning about form data handling, Batman realized he needed a way to push real-time updates to his crime monitoring dashboard. Criminals don't wait for Batman to refresh his browser!

He discovered Server-Sent Events (SSE) - a perfect solution for one-way communication from server to client over HTTP. SSE allows Batman to stream live data to his dashboard without the complexity of full bidirectional communication.

"This is exactly what I need for my crime alerts!" Batman exclaimed. "I can push updates to the dashboard instantly when new crimes are detected."

Server-Sent Events are ideal for:

Real-time notifications
Live data feeds
Progress updates
Chat applications (server-to-client only)
Dashboard updates
Log streaming

How does it work?

Batman can create Server-Sent Events streams by using the SSE_Response and SSE_Message classes. He can use both regular generators and async generators depending on his needs:

Regular generators: Perfect for simple data streams or when working with synchronous operations
Async generators: Ideal when Batman needs to perform async operations like database queries or API calls within the stream

SSE Response

GET

/events

from robyn import Robyn, SSE_Response, SSE_Message
import time

app = Robyn(__file__)

@app.get("/events")
def stream_events(request):
    def event_generator():
        for i in range(10):
            yield SSE_Message(f"Event {i}", id=str(i))
            time.sleep(1)
    
    return SSE_Response(event_generator())

Async Generators

When Batman needs to perform async operations during his SSE streams - like fetching data from databases or making API calls - he uses async generators with async def and await. This allows him to handle multiple streams concurrently without blocking other operations.

The key difference is using async def for the generator function and await for async operations inside the generator:

Advanced Async SSE

GET

/events/database

from robyn import Robyn, SSE_Response, SSE_Message
import asyncio
import json
import time

app = Robyn(__file__)

@app.get("/events/database")
async def stream_database_events(request):
    async def database_event_generator():
        for i in range(10):
            # Simulate async database query
            await asyncio.sleep(0.3)
            
            # Simulate fetching data from database
            data = {
                "crime_id": i,
                "location": f"Gotham District {i}",
                "severity": "high" if i % 2 == 0 else "low",
                "timestamp": time.time()
            }
            
            yield SSE_Message(
                json.dumps(data),
                event="crime_alert",
                id=str(i)
            )
    
    return SSE_Response(database_event_generator())

What's next?

Batman has mastered Server-Sent Events and can now stream real-time updates to his crime dashboard. While SSE is perfect for one-way communication from server to client, Batman realizes he needs bidirectional communication for more interactive features like real-time chat with his allies.

Next, he wants to explore how to handle bidirectional communication with WebSockets for more interactive features.

If Batman needs to handle unexpected situations, he'll learn about Exception handling to make his applications more robust.

For scaling his crime monitoring system across multiple processes, Batman will explore Scaling the Application.