useSSEStream

useSSEStream은 Server-Sent Events(SSE)를 통한 실시간 데이터 스트리밍을 관리하는 훅입니다. 자동 재연결, 타입 안전한 이벤트 핸들러, 연결 상태 추적 기능을 제공합니다.

핵심 기능

실시간 스트리밍

서버에서 클라이언트로단방향 실시간 데이터

자동 재연결

연결 실패 시 자동 재시도설정 가능한 횟수/간격

타입 안전

이벤트별 타입 지정제네릭 기반

연결 상태

isConnected, error 추적UI 피드백 가능

SSE란?

Server-Sent Events는 서버에서 클라이언트로 실시간 데이터를 단방향으로 전송하는 기술입니다.

WebSocket vs SSE

특징	WebSocket	SSE
통신 방향	양방향 (Full-duplex)	단방향 (Server → Client)
프로토콜	WS/WSS	HTTP/HTTPS
재연결	수동 구현 필요	브라우저 자동 재연결
복잡도	높음	낮음
사용 케이스	채팅, 게임	알림, 로그, 진행 상황

SSE를 사용해야 하는 경우

서버에서 클라이언트로만 데이터를 보내는 경우
AI 응답 스트리밍 (ChatGPT 스타일)
실시간 로그 모니터링
파일 업로드/처리 진행 상황
서버 이벤트 알림

기본 사용법

Import

import { useSSEStream } from "./sonamu.shared";

useSSEStream은 프로젝트 스캐폴딩 시 생성되는 sonamu.shared.ts에 정의되어 있습니다. services.generated.ts에서는 각 @stream API에 대한 래퍼 함수를 자동 생성하며, 내부적으로 useSSEStream을 사용합니다.

타입 정의

// 이벤트 타입 정의
type AIStreamEvents = {
  token: { text: string };     // token 이벤트: 텍스트 조각
  usage: { tokens: number };   // usage 이벤트: 토큰 사용량
  end: string;                  // end 이벤트: 완료 신호
};

기본 사용

import { useState } from "react";
import { useSSEStream } from "./sonamu.shared";

export function AIChat() {
  const [response, setResponse] = useState("");
  const [totalTokens, setTotalTokens] = useState(0);

  const { isConnected, error, isEnded } = useSSEStream<AIStreamEvents>(
    "/api/ai/stream",  // SSE 엔드포인트
    { prompt: "Hello, AI!" },  // 쿼리 파라미터
    {
      // 이벤트 핸들러
      token: (data) => {
        // 텍스트 조각이 올 때마다 추가
        setResponse((prev) => prev + data.text);
      },
      usage: (data) => {
        // 토큰 사용량 업데이트
        setTotalTokens(data.tokens);
      },
      end: () => {
        // 스트리밍 완료
        console.log("Stream ended");
      },
    },
    {
      enabled: true,      // 자동 연결
      retry: 3,           // 최대 3번 재시도
      retryInterval: 3000, // 3초 간격
    }
  );

  return (
    <div>
      {isConnected && <div className="text-green-500">연결됨</div>}
      {error && <div className="text-red-500">에러: {error}</div>}
      <div className="whitespace-pre-wrap">{response}</div>
      {isEnded && <div>토큰 사용: {totalTokens}</div>}
    </div>
  );
}

API 레퍼런스

function useSSEStream<T extends Record<string, any>>(
  url: string,
  params: Record<string, any>,
  handlers: {
    [K in keyof T]?: (data: T[K]) => void;
  },
  options?: SSEStreamOptions
): SSEStreamState

파라미터

url

SSE 엔드포인트 URL입니다.

useSSEStream("/api/ai/stream", params, handlers);

params

URL 쿼리 파라미터로 전달될 객체입니다.

useSSEStream(
  "/api/ai/stream",
  {
    prompt: "Hello",
    model: "gpt-4",
    temperature: 0.7,
  },
  handlers
);

// 실제 요청 URL:
// /api/ai/stream?prompt=Hello&model=gpt-4&temperature=0.7

handlers

이벤트별 핸들러 함수입니다.

{
  token: (data: { text: string }) => {
    // token 이벤트 처리
  },
  usage: (data: { tokens: number }) => {
    // usage 이벤트 처리
  },
  end: () => {
    // end 이벤트 처리
  },
}

특수 이벤트:

end: 서버가 명시적으로 스트림 종료를 알릴 때
message: 이벤트 타입이 지정되지 않은 기본 메시지

options

옵션	타입	기본값	설명
`enabled`	`boolean`	`true`	자동 연결 여부
`retry`	`number`	`3`	최대 재시도 횟수
`retryInterval`	`number`	`3000`	재시도 간격 (ms)

반환값 (SSEStreamState)

type SSEStreamState = {
  isConnected: boolean;  // 현재 연결 상태
  error: string | null;  // 에러 메시지
  retryCount: number;    // 현재 재시도 횟수
  isEnded: boolean;      // 정상 종료 여부
};

실전 예제

AI 응답 스트리밍

import { useState } from "react";
import { useSSEStream } from "./sonamu.shared";
import { Button, Textarea } from "@sonamu-kit/react-components/components";

type AIStreamEvents = {
  token: { text: string };
  end: string;
};

export function AIChat() {
  const [prompt, setPrompt] = useState("");
  const [response, setResponse] = useState("");
  const [enabled, setEnabled] = useState(false);

  const { isConnected, error, isEnded } = useSSEStream<AIStreamEvents>(
    "/api/ai/chat",
    { prompt },
    {
      token: (data) => {
        setResponse((prev) => prev + data.text);
      },
      end: () => {
        console.log("AI 응답 완료");
        setEnabled(false);  // 연결 종료
      },
    },
    {
      enabled,
      retry: 1,  // AI 요청은 재시도 1회만
      retryInterval: 1000,
    }
  );

  const handleSubmit = () => {
    setResponse("");  // 이전 응답 초기화
    setEnabled(true);  // 스트림 시작
  };

  return (
    <div className="space-y-4">
      <Textarea
        value={prompt}
        onValueChange={setPrompt}
        placeholder="질문을 입력하세요"
        rows={3}
      />

      <Button
        onClick={handleSubmit}
        disabled={isConnected || !prompt}
      >
        {isConnected ? "응답 중..." : "질문하기"}
      </Button>

      {error && (
        <div className="p-4 bg-red-50 text-red-700 rounded">
          {error}
        </div>
      )}

      {response && (
        <div className="p-4 bg-gray-50 rounded">
          <div className="whitespace-pre-wrap">{response}</div>
          {isConnected && <span className="animate-pulse">▊</span>}
        </div>
      )}

      {isEnded && (
        <div className="text-sm text-gray-500">
          응답 완료
        </div>
      )}
    </div>
  );
}

파일 업로드 진행 상황

type UploadEvents = {
  progress: { percent: number; loaded: number; total: number };
  complete: { fileUrl: string };
  error: { message: string };
};

export function FileUploader() {
  const [file, setFile] = useState<File | null>(null);
  const [uploadId, setUploadId] = useState<string | null>(null);
  const [progress, setProgress] = useState(0);
  const [fileUrl, setFileUrl] = useState<string | null>(null);

  const { isConnected, error } = useSSEStream<UploadEvents>(
    "/api/file/upload/progress",
    { uploadId: uploadId ?? "" },
    {
      progress: (data) => {
        setProgress(data.percent);
      },
      complete: (data) => {
        setFileUrl(data.fileUrl);
      },
      error: (data) => {
        alert(data.message);
      },
    },
    {
      enabled: uploadId !== null,
      retry: 5,
      retryInterval: 2000,
    }
  );

  const handleUpload = async () => {
    if (!file) return;

    // 1. 업로드 시작 (uploadId 발급)
    const formData = new FormData();
    formData.append("file", file);

    const response = await fetch("/api/file/upload", {
      method: "POST",
      body: formData,
    });

    const { uploadId } = await response.json();

    // 2. SSE로 진행 상황 추적
    setUploadId(uploadId);
  };

  return (
    <div>
      <input
        type="file"
        onChange={(e) => setFile(e.target.files?.[0] ?? null)}
      />

      <Button onClick={handleUpload} disabled={!file || isConnected}>
        업로드
      </Button>

      {isConnected && (
        <div>
          <div className="w-full bg-gray-200 rounded-full h-2">
            <div
              className="bg-blue-600 h-2 rounded-full transition-all"
              style={{ width: `${progress}%` }}
            />
          </div>
          <p>{progress}%</p>
        </div>
      )}

      {fileUrl && (
        <div>
          업로드 완료: <a href={fileUrl}>{fileUrl}</a>
        </div>
      )}

      {error && <div className="text-red-500">{error}</div>}
    </div>
  );
}

실시간 로그 모니터링

type LogEvents = {
  log: { level: string; message: string; timestamp: string };
};

export function LogViewer() {
  const [logs, setLogs] = useState<Array<{ level: string; message: string; timestamp: string }>>([]);
  const [enabled, setEnabled] = useState(true);

  const { isConnected, error } = useSSEStream<LogEvents>(
    "/api/logs/stream",
    { level: "info" },
    {
      log: (data) => {
        setLogs((prev) => [...prev, data].slice(-100));  // 최근 100개만 유지
      },
    },
    {
      enabled,
      retry: 10,  // 로그는 계속 재연결
      retryInterval: 5000,
    }
  );

  return (
    <div>
      <div className="flex items-center gap-2 mb-4">
        <Button
          variant={enabled ? "red" : "default"}
          onClick={() => setEnabled(!enabled)}
        >
          {enabled ? "중지" : "시작"}
        </Button>

        {isConnected && <span className="text-green-500">● 연결됨</span>}
        {error && <span className="text-red-500">● {error}</span>}
      </div>

      <div className="bg-black text-green-400 p-4 rounded font-mono text-sm h-96 overflow-y-auto">
        {logs.map((log, i) => (
          <div key={i} className="mb-1">
            <span className="text-gray-500">[{log.timestamp}]</span>{" "}
            <span className={log.level === "error" ? "text-red-400" : ""}>
              {log.level.toUpperCase()}
            </span>{" "}
            {log.message}
          </div>
        ))}
      </div>
    </div>
  );
}

백엔드 구현

SSE 엔드포인트는 @stream 데코레이터와 ctx.createSSE()로 만듭니다.

기본 SSE API

// ai.model.ts (백엔드)
import { BaseModelClass, stream, Sonamu, z } from "sonamu";

class AIModelClass extends BaseModelClass {
  @stream({
    type: 'sse',
    events: z.object({
      token: z.object({ text: z.string() }),
      usage: z.object({ tokens: z.number() }),
    }),
    guards: ["login"],
  })
  async chat(prompt: string): Promise<void> {
    const ctx = Sonamu.getContext();
    const sse = ctx.createSSE(
      z.object({
        token: z.object({ text: z.string() }),
        usage: z.object({ tokens: z.number() }),
      })
    );

    const stream = await openai.chat.completions.create({
      model: "gpt-4",
      messages: [{ role: "user", content: prompt }],
      stream: true,
    });

    for await (const chunk of stream) {
      const text = chunk.choices[0]?.delta?.content ?? "";
      if (text) {
        sse.publish('token', { text });
      }
    }

    sse.publish('usage', { tokens: 42 });
    await sse.end();
  }
}

publish 형식

sse.publish('이벤트명', { /* 데이터 */ });

예시:

sse.publish('token', { text: "Hello" });
sse.publish('usage', { tokens: 42 });
await sse.end();  // 스트림 종료 시 end 이벤트 전송

SSE 엔드포인트 구현에 대한 자세한 내용은 SSE 엔드포인트 만들기를 참고하세요.

고급 기능

조건부 연결

const [shouldConnect, setShouldConnect] = useState(false);

const { isConnected } = useSSEStream(
  "/api/notifications",
  {},
  handlers,
  {
    enabled: shouldConnect,  // false면 연결하지 않음
  }
);

// 사용자 액션으로 연결 시작
<Button onClick={() => setShouldConnect(true)}>
  알림 구독
</Button>

동적 파라미터

const [filter, setFilter] = useState("all");

useSSEStream(
  "/api/logs/stream",
  { level: filter },  // filter가 바뀌면 재연결
  handlers
);

// filter가 변경되면 자동으로 연결이 끊기고 새 파라미터로 재연결됨

수동 재연결

const [reconnectTrigger, setReconnectTrigger] = useState(0);

useSSEStream(
  "/api/stream",
  { trigger: reconnectTrigger },  // trigger 변경 시 재연결
  handlers
);

// 수동 재연결
<Button onClick={() => setReconnectTrigger(prev => prev + 1)}>
  재연결
</Button>

여러 스트림 동시 사용

// 스트림 1: AI 응답
const stream1 = useSSEStream<AIEvents>(
  "/api/ai/chat",
  { prompt: "Hello" },
  {
    token: (data) => setAiResponse(prev => prev + data.text),
  }
);

// 스트림 2: 알림
const stream2 = useSSEStream<NotificationEvents>(
  "/api/notifications",
  {},
  {
    notification: (data) => showNotification(data),
  }
);

주의사항

1. EventSource 제한

HTTP/2에서는 브라우저당 최대 6개 연결 제한
너무 많은 SSE 연결을 동시에 열지 마세요

2. 메모리 관리

스트리밍 데이터를 무한정 누적하지 마세요.

// ❌ 나쁜 예: 메모리 무한 증가
const [logs, setLogs] = useState<string[]>([]);

useSSEStream("/api/logs", {}, {
  log: (data) => {
    setLogs(prev => [...prev, data.message]);  // 계속 증가
  },
});

// ✅ 좋은 예: 최근 N개만 유지
useSSEStream("/api/logs", {}, {
  log: (data) => {
    setLogs(prev => [...prev, data.message].slice(-100));  // 최근 100개만
  },
});

3. 에러 처리

const { error, isEnded } = useSSEStream(url, params, {
  error: (data) => {
    // 서버에서 보낸 에러 이벤트
    console.error("Server error:", data);
  },
});

// 연결 실패 에러
if (error) {
  console.error("Connection error:", error);
}

4. 컴포넌트 언마운트

컴포넌트가 언마운트되면 자동으로 연결이 닫힙니다.

useEffect(() => {
  // useSSEStream은 자동으로 cleanup 처리됨
  return () => {
    // 별도 cleanup 불필요
  };
}, []);

문제 해결

SSE가 연결되지 않음

원인: CORS 설정 또는 엔드포인트 오류 해결:

// 백엔드: sonamu.config.ts
export default {
  api: {
    cors: {
      origin: "http://localhost:5173",
      credentials: true,
    },
  },
};

재연결이 계속 실패함

원인: 백엔드 서버가 다운되었거나 네트워크 문제 해결: retry와 retryInterval을 조정하거나, 에러 UI를 표시하세요.

데이터가 파싱되지 않음

원인: 서버에서 JSON이 아닌 형식으로 데이터 전송 해결: 백엔드에서 sse.publish('이벤트명', JSON객체) 형식을 사용하세요.

​핵심 기능

실시간 스트리밍

자동 재연결

타입 안전

연결 상태

​SSE란?

​WebSocket vs SSE

​기본 사용법

​Import

​타입 정의

​기본 사용

​API 레퍼런스

​useSSEStream

​파라미터

​url

​params

​handlers

​options

​반환값 (SSEStreamState)

​실전 예제

​AI 응답 스트리밍

​파일 업로드 진행 상황

​실시간 로그 모니터링

​백엔드 구현

​기본 SSE API

​publish 형식

​고급 기능

​조건부 연결

​동적 파라미터

​수동 재연결

​여러 스트림 동시 사용

​주의사항

​1. EventSource 제한

​2. 메모리 관리

​3. 에러 처리

​4. 컴포넌트 언마운트

​문제 해결

​SSE가 연결되지 않음

​재연결이 계속 실패함

​데이터가 파싱되지 않음

​관련 문서

핵심 기능

SSE란?

WebSocket vs SSE

기본 사용법

Import

타입 정의

기본 사용

API 레퍼런스

useSSEStream

파라미터

url

params

handlers

options

반환값 (SSEStreamState)

실전 예제

AI 응답 스트리밍

파일 업로드 진행 상황

실시간 로그 모니터링

백엔드 구현

기본 SSE API

publish 형식

고급 기능

조건부 연결

동적 파라미터

수동 재연결

여러 스트림 동시 사용

주의사항

1. EventSource 제한

2. 메모리 관리

3. 에러 처리

4. 컴포넌트 언마운트

문제 해결

SSE가 연결되지 않음

재연결이 계속 실패함

데이터가 파싱되지 않음

관련 문서