SSE

// Express.js SSE 서버
const express = require('express');
const app = express();

app.get('/api/events', (req, res) => {
    // SSE 헤더 설정
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');
    res.setHeader('Access-Control-Allow-Origin', '*');

    // 연결 유지를 위한 heartbeat
    const heartbeat = setInterval(() => {
        res.write(':heartbeat\n\n');
    }, 30000);

    // 클라이언트 ID 추적
    const clientId = Date.now();
    console.log(`클라이언트 연결: ${clientId}`);

    // 데이터 전송 함수
    function sendEvent(data, eventType = null, id = null) {
        if (id) res.write(`id: ${id}\n`);
        if (eventType) res.write(`event: ${eventType}\n`);
        res.write(`data: ${JSON.stringify(data)}\n\n`);
    }

    // 초기 데이터 전송
    sendEvent({ message: '연결 성공!', clientId }, 'connected');

    // 예시: 1초마다 시간 전송
    const interval = setInterval(() => {
        sendEvent({
            time: new Date().toISOString(),
            random: Math.random()
        });
    }, 1000);

    // 연결 종료 처리
    req.on('close', () => {
        clearInterval(interval);
        clearInterval(heartbeat);
        console.log(`클라이언트 연결 해제: ${clientId}`);
    });
});

// 여러 클라이언트에게 브로드캐스트
const clients = new Set();

app.get('/api/stream', (req, res) => {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');

    clients.add(res);
    req.on('close', () => clients.delete(res));
});

function broadcast(data) {
    clients.forEach(client => {
        client.write(`data: ${JSON.stringify(data)}\n\n`);
    });
}

app.listen(3000);

# FastAPI SSE 서버
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
import asyncio
import json

app = FastAPI()

# 간단한 SSE 엔드포인트
@app.get("/api/events")
async def events():
    async def event_generator():
        while True:
            # 데이터 생성
            data = {
                "timestamp": str(asyncio.get_event_loop().time()),
                "message": "Hello SSE!"
            }
            yield f"data: {json.dumps(data)}\n\n"
            await asyncio.sleep(1)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

# sse-starlette 라이브러리 사용 (권장)
# pip install sse-starlette

@app.get("/api/notifications")
async def notifications(request: Request):
    async def event_stream():
        message_id = 0
        while True:
            if await request.is_disconnected():
                break

            message_id += 1
            yield {
                "event": "notification",
                "id": str(message_id),
                "retry": 5000,  # 재연결 대기 시간 (ms)
                "data": json.dumps({
                    "title": "새 알림",
                    "body": f"메시지 #{message_id}"
                })
            }
            await asyncio.sleep(2)

    return EventSourceResponse(event_stream())

# ChatGPT 스타일 AI 스트리밍 응답
@app.get("/api/chat")
async def chat_stream(prompt: str):
    async def generate_response():
        # AI 모델 응답 시뮬레이션
        response = "안녕하세요! SSE를 사용한 스트리밍 응답입니다."
        for char in response:
            yield f"data: {json.dumps({'token': char})}\n\n"
            await asyncio.sleep(0.05)  # 타이핑 효과
        yield f"data: {json.dumps({'done': True})}\n\n"

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