通信プロトコル「SSE」を理解する

id: 42
event: priceUpdate
data: {"symbol":"BTC","price":8500000}
retry: 3000
 
data: シンプルなメッセージ（event フィールドなし）
 
: これはコメント行（クライアントには届かない）

sequenceDiagram
    participant C as Client (EventSource)
    participant S as Server
 
    C->>S: GET /events<br/>Accept: text/event-stream
    S-->>C: 200 OK<br/>Content-Type: text/event-stream
 
    loop 接続を維持したまま継続
        S-->>C: id: 1\ndata: event1\n\n
        S-->>C: id: 2\ndata: event2\n\n
        S-->>C: : keepalive（コメント）
    end
 
    note over C,S: ネットワーク障害などで切断
 
    C->>S: GET /events<br/>Last-Event-ID: 2（自動再接続）
    S-->>C: 200 OK（id: 2 以降のイベントから再開）

sequenceDiagram
    participant U as ユーザー (Browser)
    participant API as API Server
    participant LLM as LLM Engine
 
    U->>API: POST /chat<br/>{"message": "こんにちは"}
    API->>LLM: プロンプト送信（推論開始）
 
    API-->>U: 200 OK<br/>Content-Type: text/event-stream
 
    loop トークンを逐次ストリーミング
        LLM-->>API: token: "こん"
        API-->>U: data: {"token":"こん","done":false}
 
        LLM-->>API: token: "にちは"
        API-->>U: data: {"token":"にちは","done":false}
 
        LLM-->>API: token: "！"
        API-->>U: data: {"token":"！","done":false}
    end
 
    LLM-->>API: 生成完了
    API-->>U: data: {"token":"","done":true}

event: message_start
data: {"type":"message_start","message":{"id":"msg_01...","role":"assistant",...}}
 
event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"こん"}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"にちは"}}
 
event: content_block_stop
data: {"type":"content_block_stop","index":0}
 
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}
 
event: message_stop
data: {"type":"message_stop"}

package main
 
import (
    "bufio"
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "strings"
)
 
// ClaudeStreamHandler はクライアントへ Claude のストリームをそのまま中継する
func ClaudeStreamHandler(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "SSE not supported", http.StatusInternalServerError)
        return
    }
 
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    w.Header().Set("Access-Control-Allow-Origin", "*")
 
    // リクエストボディからユーザーメッセージを取得
    var req struct {
        Message string `json:"message"`
    }
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, "invalid request", http.StatusBadRequest)
        return
    }
 
    // Anthropic API へストリーミングリクエスト
    payload := map[string]any{
        "model":      "claude-sonnet-4.6",
        "max_tokens": 1024,
        "stream":     true,
        "messages": []map[string]string{
            {"role": "user", "content": req.Message},
        },
    }
    body, _ := json.Marshal(payload)
 
    apiReq, _ := http.NewRequestWithContext(r.Context(), "POST",
        "https://api.anthropic.com/v1/messages", bytes.NewReader(body))
    apiReq.Header.Set("x-api-key", "YOUR_API_KEY")
    apiReq.Header.Set("anthropic-version", "2023-06-01")
    apiReq.Header.Set("content-type", "application/json")
 
    resp, err := http.DefaultClient.Do(apiReq)
    if err != nil {
        fmt.Fprintf(w, "event: error\ndata: {\"message\":\"%v\"}\n\n", err)
        flusher.Flush()
        return
    }
    defer resp.Body.Close()
 
    // SSE をそのままクライアントへ中継
    scanner := bufio.NewScanner(resp.Body)
    for scanner.Scan() {
        line := scanner.Text()
 
        // content_block_delta からテキストだけ抽出して転送する場合
        if strings.HasPrefix(line, "data: ") {
            fmt.Fprintf(w, "%s\n\n", line)
            flusher.Flush()
        }
 
        // クライアント切断を検知して終了
        select {
        case <-r.Context().Done():
            return
        default:
        }
    }
}

stateDiagram-v2
    [*] --> Idle
    Idle --> Connecting : POST /chat 送信
    Connecting --> Streaming : 200 OK 受信
    Streaming --> Streaming : content_block_delta 受信 トークンを UI に追加
    Streaming --> Done : message_stop 受信
    Streaming --> Error : ネットワーク切断 / エラー
    Error --> Connecting : 自動再接続（Last-Event-ID）
    Done --> Idle : 次のメッセージ待ち

package main
 
import (
    "fmt"
    "math/rand"
    "net/http"
    "time"
)
 
func sseHandler(w http.ResponseWriter, r *http.Request) {
    // SSE に必要なヘッダーを設定
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    // CORS が必要な場合
    w.Header().Set("Access-Control-Allow-Origin", "*")
 
    // http.Flusher を取得（チャンク送信に必須）
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "SSE not supported", http.StatusInternalServerError)
        return
    }
 
    // クライアント切断を検知
    ctx := r.Context()
    eventID := 1
 
    for {
        select {
        case <-ctx.Done(): // クライアントが切断した
            return
        default:
        }
 
        // イベントを送信
        price := 8000000 + rand.Intn(500000)
        fmt.Fprintf(w, "id: %d\n", eventID)
        fmt.Fprintf(w, "event: priceUpdate\n")
        fmt.Fprintf(w, "data: {\"price\":%d}\n", price)
        fmt.Fprintf(w, "\n") // イベント終端の空行
        flusher.Flush()      // バッファをフラッシュ（重要！）
 
        eventID++
        time.Sleep(1 * time.Second)
    }
}
 
func main() {
    http.HandleFunc("/events", sseHandler)
    fmt.Println("Server running on :8080")
    http.ListenAndServe(":8080", nil)
}

graph LR
    SRC["イベントソース\n(DB / Queue / Timer)"]
 
    subgraph Server["Go Server"]
        B["Broker\n(goroutine)"]
        H1["Handler\ngoroutine"]
        H2["Handler\ngoroutine"]
        H3["Handler\ngoroutine"]
    end
 
    C1["Client 1\nEventSource"]
    C2["Client 2\nEventSource"]
    C3["Client 3\nEventSource"]
 
    SRC -->|broadcast chan| B
    B -->|chan string| H1
    B -->|chan string| H2
    B -->|chan string| H3
    H1 -->|SSE stream| C1
    H2 -->|SSE stream| C2
    H3 -->|SSE stream| C3

package main
 
import (
    "fmt"
    "net/http"
    "sync"
)
 
// Broker はクライアントへのブロードキャストを管理する
type Broker struct {
    clients   map[chan string]struct{}
    mu        sync.RWMutex
    subscribe chan chan string
    unsub     chan chan string
    broadcast chan string
}
 
func NewBroker() *Broker {
    b := &Broker{
        clients:   make(map[chan string]struct{}),
        subscribe: make(chan chan string),
        unsub:     make(chan chan string),
        broadcast: make(chan string, 100),
    }
    go b.run()
    return b
}
 
func (b *Broker) run() {
    for {
        select {
        case client := <-b.subscribe:
            b.clients[client] = struct{}{}
 
        case client := <-b.unsub:
            delete(b.clients, client)
            close(client)
 
        case msg := <-b.broadcast:
            for client := range b.clients {
                select {
                case client <- msg: // ノンブロッキング送信
                default:            // 遅延クライアントはスキップ
                }
            }
        }
    }
}
 
func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "SSE not supported", http.StatusInternalServerError)
        return
    }
 
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
 
    client := make(chan string, 10)
    b.subscribe <- client
    defer func() { b.unsub <- client }()
 
    for {
        select {
        case <-r.Context().Done():
            return
        case msg := <-client:
            fmt.Fprintf(w, "data: %s\n\n", msg)
            flusher.Flush()
        }
    }
}

// セマフォで同時接続数を制限
type Server struct {
    sem chan struct{} // バッファサイズ = 最大同時接続数
}
 
func NewServer(maxConns int) *Server {
    return &Server{sem: make(chan struct{}, maxConns)}
}
 
func (s *Server) Handler(w http.ResponseWriter, r *http.Request) {
    select {
    case s.sem <- struct{}{}: // 接続枠を取得
        defer func() { <-s.sem }() // 終了時に解放
    default:
        http.Error(w, "Too many connections", http.StatusTooManyRequests)
        return
    }
    // ... SSE 処理
}

// ❌ EventSource：ヘッダー設定不可・GET しか使えない
new EventSource('/events', { headers: { 'Authorization': 'Bearer token' } });

sequenceDiagram
    participant C as Client (SPA)
    participant API as API Server
    participant LLM as LLM Engine
 
    C->>API: POST /auth { email, password }
    API-->>C: 200 OK { token: "eyJ..." }
 
    C->>API: POST /chat<br/>Authorization: Bearer eyJ...<br/>body: { message: "..." }
    API->>LLM: プロンプト送信
    API-->>C: 200 OK（SSE ストリーム開始）
 
    loop トークン逐次配信
        LLM-->>API: token
        API-->>C: data: {"token":"..."}
    end
 
    API-->>C: data: {"done":true}