2장: SSE(Server-Sent Events) 심층 분석

event: message
data: 첫 번째 토큰입니다
id: 1
 
event: message
data: 두 번째 토큰입니다
id: 2
 
event: done
data: [DONE]

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

// 기본 사용법
const source = new EventSource("/api/stream");
 
// 기본 message 이벤트 수신
source.onmessage = (event: MessageEvent) => {
  const token = event.data;
  appendToUI(token);
};
 
// 커스텀 이벤트 수신
source.addEventListener("reasoning", (event: MessageEvent) => {
  showReasoningStep(event.data);
});
 
// 연결 상태 확인
// 0: CONNECTING, 1: OPEN, 2: CLOSED
console.log(source.readyState);
 
// 에러 처리
source.onerror = (event: Event) => {
  if (source.readyState === EventSource.CLOSED) {
    console.log("연결이 종료되었습니다");
  }
};
 
// 명시적 연결 종료
source.close();

async function streamChat(message: string) {
  const response = await fetch("/api/chat", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${token}`,
    },
    body: JSON.stringify({ message }),
  });
 
  if (!response.ok) {
    throw new Error(`HTTP ${response.status}`);
  }
 
  const reader = response.body?.getReader();
  const decoder = new TextDecoder();
 
  if (!reader) throw new Error("ReadableStream not supported");
 
  let buffer = "";
 
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
 
    buffer += decoder.decode(value, { stream: true });
 
    // SSE 형식 파싱: 빈 줄로 메시지 구분
    const messages = buffer.split("\n\n");
    buffer = messages.pop() ?? "";
 
    for (const msg of messages) {
      const dataLine = msg
        .split("\n")
        .find((line) => line.startsWith("data: "));
      if (dataLine) {
        const data = dataLine.slice(6);
        if (data === "[DONE]") return;
        appendToken(data);
      }
    }
  }
}

import { NextRequest } from "next/server";
 
export async function POST(request: NextRequest) {
  const { message } = await request.json();
 
  const encoder = new TextEncoder();
 
  const stream = new ReadableStream({
    async start(controller) {
      try {
        // LLM API 호출 (예: OpenAI)
        const response = await fetch(
          "https://api.openai.com/v1/chat/completions",
          {
            method: "POST",
            headers: {
              "Content-Type": "application/json",
              Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
            },
            body: JSON.stringify({
              model: "gpt-4o",
              messages: [{ role: "user", content: message }],
              stream: true,
            }),
          }
        );
 
        const reader = response.body?.getReader();
        if (!reader) throw new Error("No reader");
 
        const decoder = new TextDecoder();
        let buffer = "";
 
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
 
          buffer += decoder.decode(value, { stream: true });
          const lines = buffer.split("\n");
          buffer = lines.pop() ?? "";
 
          for (const line of lines) {
            if (line.startsWith("data: ")) {
              const data = line.slice(6);
              if (data === "[DONE]") {
                controller.enqueue(
                  encoder.encode("data: [DONE]\n\n")
                );
                controller.close();
                return;
              }
 
              const parsed = JSON.parse(data);
              const token =
                parsed.choices[0]?.delta?.content ?? "";
              if (token) {
                controller.enqueue(
                  encoder.encode(`data: ${token}\n\n`)
                );
              }
            }
          }
        }
      } catch (error) {
        controller.enqueue(
          encoder.encode(
            `event: error\ndata: ${JSON.stringify({
              message: "스트리밍 중 오류 발생",
            })}\n\n`
          )
        );
        controller.close();
      }
    },
  });
 
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    },
  });
}

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from anthropic import Anthropic
import json
import asyncio
 
app = FastAPI()
client = Anthropic()
 
 
async def generate_stream(message: str):
    """Anthropic Claude 스트리밍 응답 생성기"""
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}],
    ) as stream:
        for text in stream.text_stream:
            # SSE 형식으로 인코딩
            data = json.dumps({"token": text}, ensure_ascii=False)
            yield f"data: {data}\n\n"
            await asyncio.sleep(0)  # 이벤트 루프에 제어권 반환
 
    yield "data: [DONE]\n\n"
 
 
@app.post("/api/chat")
async def chat(request: dict):
    message = request.get("message", "")
 
    return StreamingResponse(
        generate_stream(message),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Nginx 버퍼링 비활성화
        },
    )

// SSE에서 생성 중단 패턴
const abortController = new AbortController();
 
// SSE 스트리밍 시작
fetch("/api/chat", { signal: abortController.signal });
 
// 중단 버튼 클릭 시
function handleStop() {
  abortController.abort(); // 연결 종료
  // 또는 서버에 별도 요청
  fetch("/api/chat/cancel", { method: "POST" });
}