SSE · GitHub

You:

write me an example server sent events code

ChatGPT:

Server-Sent Events (SSE) is a mechanism for a server to push real-time updates to a web client over an HTTP connection. Here's an example demonstrating a basic implementation:

Server-side (Node.js Example)

// Import required modules
const http = require('http');

// Create an HTTP server
const server = http.createServer((req, res) => {
    // Check if the request is for the SSE endpoint
    if (req.url === '/events') {
        // Set the appropriate headers for SSE
        res.writeHead(200, {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
        });

        // Send an initial message
        res.write(`data: Welcome to SSE server\n\n`);

        // Send periodic updates every 2 seconds
        const interval = setInterval(() => {
            const message = JSON.stringify({ timestamp: new Date().toISOString() });
            res.write(`data: ${message}\n\n`);
        }, 2000);

        // Handle client disconnect
        req.on('close', () => {
            clearInterval(interval);
            console.log('Client disconnected');
        });
    } else {
        // Fallback for other routes
        res.writeHead(404, { 'Content-Type': 'text/plain' });
        res.end('Not Found');
    }
});

// Start the server
server.listen(3000, () => {
    console.log('SSE server running at http://localhost:3000/events');
});

You:

use python fastapi

ChatGPT:

Here’s how to implement Server-Sent Events (SSE) using FastAPI in Python:

Install Required Libraries

First, install FastAPI and an ASGI server like uvicorn:

pip install fastapi uvicorn

FastAPI SSE Implementation

Below is an example of a FastAPI application that streams events to the client.

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import datetime

app = FastAPI()

# Generator function to produce event stream
async def event_stream():
    while True:
        # Generate a new event with a timestamp
        data = {"timestamp": datetime.datetime.now().isoformat()}
        yield f"data: {data}\n\n"
        await asyncio.sleep(2)  # Send an event every 2 seconds

# Route for Server-Sent Events
@app.get("/events")
async def sse_endpoint(request: Request):
    # Stream the response
    async def generator():
        try:
            async for event in event_stream():
                if await request.is_disconnected():
                    break
                yield event
        except asyncio.CancelledError:
            pass

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

# Root route for testing
@app.get("/")
async def root():
    return {"message": "Go to /events to see the SSE in action"}

Running the Server

Run the application with uvicorn:

uvicorn app:app --reload

Client-Side Example

To consume the SSE, you can use JavaScript in the browser:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SSE Client</title>
</head>
<body>
    <h1>Server-Sent Events Example</h1>
    <div id="events"></div>
    <script>
        const eventSource = new EventSource("/events");

        eventSource.onmessage = (event) => {
            const data = JSON.parse(event.data);
            const div = document.getElementById("events");
            div.innerHTML += `<p>New Event: ${data.timestamp}</p>`;
        };

        eventSource.onerror = (err) => {
            console.error("EventSource failed:", err);
            eventSource.close();
        };
    </script>
</body>
</html>

Testing

Run the FastAPI server (uvicorn app:app --reload).
Open the browser and navigate to /events to verify SSE is streaming events.
Optionally, serve the HTML file to view client-side consumption.

This code sets up a working SSE server with FastAPI and a basic browser client.