How to build a SSE stream api to frontend

Server-Sent Events (SSE) 是一種基於 HTTP 的技術，允許伺服器持續向客戶端發送事件，主要用於實時應用。以下是 SSE 的一些關鍵特點和使用情境：

基本概念

單向通訊：SSE 允許伺服器向客戶端推送數據，但客戶端無法主動向伺服器發送數據。這意味著如果客戶端需要與伺服器通訊，必須通過 API 調用來實現，這使得實現上相對繁瑣。
HTTP 協議：SSE 使用 HTTP 協議進行通訊，這使得它在瀏覽器的 Fetch 或 XHR 中可見。這與 WebSocket (WS) 不同，後者使用不同的協議。
內容類型：SSE 的響應內容類型為 text/event-stream，這是根據 RFC 8895 的定義來的。這個 RFC 詳細描述了 SSE 的功能和使用情境。

數據格式

鍵值對格式：SSE 的數據負載格式通常不是常見的 JSON 或 XML，而是使用鍵值對的形式。舉例來說，發送的數據可能如下所示：
```
event: start
id: 1
data: hello there
```
編碼限制：SSE 僅支持 UTF-8 編碼，這可能會對某些應用造成不便。

連線管理

連線控制：伺服器無法主動關閉連線，必須等待客戶端主動關閉，這在某些情況下可能會導致資源管理上的困難。
會話管理：由於 SSE 基於 HTTP，伺服器需要維護一個連線列表，這意味著伺服器需要管理多個會話，以便能夠正確地向每個客戶端發送數據。

與 WebSocket 的比較

特徵	SSE	WebSocket
通訊方向	單向（伺服器到客戶端）	雙向
協議	基於 HTTP	自定義協議
數據格式	鍵值對	可自定義格式（如 JSON）
連線管理	伺服器需維護連線列表	連線持久化，無需額外管理
編碼	只支持 UTF-8	可支持多種編碼

總結來說，SSE 是一種適合於需要伺服器主動推送數據的應用場景，但由於其單向通訊和 HTTP 基礎的特性，可能在某些情況下不如 WebSocket 靈活。

Citations: [1] https://datatracker.ietf.org/doc/html/rfc8895 [2] https://www.encora.com/insights/real-time-communication-simplified-a-deep-dive-into-server-sent-events-sse [3] https://stackoverflow.com/questions/63583989/performance-difference-between-websocket-and-server-sent-events-sse-for-chat-r [4] https://en.wikipedia.org/wiki/Server-sent_events [5] https://softwaremill.com/sse-vs-websockets-comparing-real-time-communication-protocols/ [6] https://ably.com/blog/websockets-vs-sse [7] https://apidog.com/blog/websockets-vs-server-sent-events/ [8] https://www.reddit.com/r/ExperiencedDevs/comments/1845vtf/websockets_vs_server_sent_events/ [9] https://www.svix.com/resources/faq/websocket-vs-sse/

SSE 的工作原理：

客戶端發送請求：客戶端通過 EventSource API 向伺服器發送一個長時間連接請求。
伺服器推送資料：伺服器持續保持連接並使用 text/event-stream 格式將資料以事件流的形式推送到客戶端。
客戶端接收資料：客戶端收到事件資料後，可以立即處理並顯示，無需再向伺服器發送請求。

簡單範例：

1. 伺服器端 (使用 Python 和 FastAPI):

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

@app.get("/sse")
async def sse_endpoint():
    def event_stream():
        for i in range(1, 6):
            yield f"data: Hello {i}\n\n"
            time.sleep(1)  # 模擬延遲

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

這段程式碼展示了一個簡單的 SSE 伺服器，該伺服器每秒推送一條消息給客戶端，共推送五次。

2. 客戶端 (HTML 和 JavaScript):

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>SSE Example</title>
</head>
<body>
    <h1>Server-Sent Events Example</h1>
    <div id="messages"></div>

    <script>
        const evtSource = new EventSource("http://localhost:8000/sse");

        evtSource.onmessage = function(event) {
            const newElement = document.createElement("div");
            newElement.textContent = `Message: ${event.data}`;
            document.getElementById("messages").appendChild(newElement);
        };

        evtSource.onerror = function(event) {
            console.error("SSE Error:", event);
            evtSource.close();  // 當出現錯誤時關閉連接
        };
    </script>
</body>
</html>

這個 HTML 頁面使用 JavaScript 的 EventSource API 與伺服器建立 SSE 連接，並將收到的每條消息顯示在頁面上。

總結：

SSE 是一種簡單且有效的伺服器向客戶端推送即時更新的方式，適合需要伺服器單向推送資料的應用場景。相比於 WebSocket，SSE 的實現更為簡單且適用於單向的即時推送需求。

StreamingResponse 是 FastAPI 提供的一種特殊的 HTTP 回應類型，用於在伺服器端生成數據並逐步發送到客戶端，而不是一次性將所有數據生成完後再發送。這在處理大量數據或需要持續發送數據的情況下特別有用，比如實現 Server-Sent Events (SSE)、逐步生成大型文件、音頻/視頻串流等。

`StreamingResponse` 的用途：

逐步生成和傳輸數據：當數據量很大或者需要實時傳輸時，StreamingResponse 可以避免一次性加載所有數據到記憶體中，而是邊生成邊傳輸，減少記憶體占用。
處理實時數據：在需要實時推送數據的應用中，例如即時通知、直播、音頻串流等，StreamingResponse 可以讓伺服器端逐步生成並發送數據。

如何使用 `StreamingResponse`：

基本範例：

假設你有一個需要逐步生成的大量數據，並希望將這些數據逐步發送給客戶端：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

def generate_large_data():
    for i in range(1, 101):
        yield f"data item {i}\n"
        time.sleep(0.1)  # 模擬數據生成的延遲

@app.get("/stream-data")
def stream_data():
    return StreamingResponse(generate_large_data(), media_type="text/plain")

說明：

generate_large_data 函數：
- 這個生成器函數模擬逐步生成數據，每次生成一個數據項，並使用 yield 關鍵字逐個返回。
- 每次生成數據後模擬延遲 0.1 秒，以展示逐步生成數據的效果。
StreamingResponse 的使用：
- 在 stream_data 路由中，StreamingResponse 被用來包裝 generate_large_data 函數。這樣，FastAPI 會將數據逐步發送給客戶端，而不是一次性傳輸所有數據。
- media_type 指定了回應的內容類型，這裡設置為 text/plain，因為我們的數據是純文本。
客戶端的行為：
- 客戶端在請求這個路徑時，不會一次性接收到所有數據，而是會隨著數據生成逐步接收並顯示。

SSE 和 `StreamingResponse` 結合使用的範例：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream():
    for i in range(1, 6):
        yield f"data: Message {i}\n\n"
        await asyncio.sleep(1)

@app.get("/sse")
async def sse():
    return StreamingResponse(event_stream(), media_type="text/event-stream")

說明：

event_stream 函數：
- 這是一個異步生成器函數，模擬了一個每秒發送一條消息的情況，適用於 SSE。
StreamingResponse 用於 SSE：
- 使用 StreamingResponse 來將每個消息作為 SSE 事件發送給客戶端。

總結：

StreamingResponse 是 FastAPI 中的一個強大工具，它允許你在數據生成時立即開始傳輸，對於處理大量數據或實時應用非常有用。當與 SSE 結合使用時，它可以讓伺服器端逐步發送實時事件給客戶端，實現實時通知等功能。

eepson123tw / public-notes