Fetch API 기반 SSE로 실시간 이벤트 스트림 구현하기

const eventSource = new EventSource('https://api.example.com/email-progress');
 
eventSource.onmessage = (event) => {
  console.log('Server says:', event.data);
};
 
eventSource.onerror = (error) => {
  console.error('SSE error:', error);
};

function parseSSEEvent(raw: string): SSEEvent | null {
  const lines = raw.split('\n');
  let eventType: string | undefined;
  let data = '';
  let id: string | undefined;
 
  for (const line of lines) {
    if (line.startsWith('event:')) {
      eventType = line.replace('event:', '').trim();
    } else if (line.startsWith('data:')) {
      // 여러 data: 줄이 있을 수 있으므로 줄바꿈으로 연결
      data += line.replace('data:', '').trim() + '\n';
    } else if (line.startsWith('id:')) {
      id = line.replace('id:', '').trim();
    }
  }
 
  if (data !== '') {
    return { event: eventType, data: data.trim(), id };
  }
  return null;
}

export class CustomSSE {
  url: string;
  onopen: ((event: Event) => void) | null = null;
  onmessage: ((event: MessageEvent) => void) | null = null;
  onerror: ((event: Event) => void) | null = null;
  readyState: number = 0; // 0: connecting, 1: open, 2: closed
 
  private _abortController: AbortController;
 
  constructor(url: string) {
    this.url = url;
    this._abortController = new AbortController();
  }
 
  private getToken(): string {
    return useAuthStore.getState().accessToken;
  }
}

async connect() {
  try {
    const response = await fetch(this.url, {
      headers: {
        Authorization: `Bearer ${this.getToken()}`,
        Accept: 'text/event-stream',            // SSE 스트림 요청
      },
      cache: 'no-store',
      credentials: 'include',
      signal: this._abortController.signal,      // 연결 취소를 위한 signal
    });
    if (!response.ok || !response.body) {
      throw new Error(`HTTP error! status: ${response.status}`);
    }
 
    // 연결 성공: readyState를 OPEN(1)으로 변경하고 onopen 콜백 호출
    this.readyState = 1;
    if (this.onopen) {
      this.onopen(new Event('open'));
    }
 
    // 서버로부터 데이터를 읽어오기 위한 리더 생성
    const reader = response.body.getReader();
    const decoder = new TextDecoder('utf-8');
    let buffer = '';
 
    // 스트림에서 데이터를 계속 읽어옵니다.
    while (true) {
      const { done, value } = await reader.read();
      if (done) break; // 더 이상 읽을 데이터가 없으면 종료
 
      // 받아온 바이트 데이터를 문자열로 변환하고, 버퍼에 누적합니다.
      buffer += decoder.decode(value, { stream: true });
 
      // 버퍼 내에서 "\n\n" 구분자로 이벤트 단위를 찾습니다.
      let delimiterIndex: number;
      while ((delimiterIndex = buffer.indexOf('\n\n')) !== -1) {
        const rawEvent = buffer.slice(0, delimiterIndex).trim();
        buffer = buffer.slice(delimiterIndex + 2);
        if (rawEvent) {
          // parseSSEEvent 함수를 통해 원시 이벤트 텍스트를 파싱
          const eventData = parseSSEEvent(rawEvent);
          if (eventData && this.onmessage) {
            // 파싱된 데이터를 MessageEvent 형태로 변환하여 onmessage 콜백에 전달
            const messageEvent = new MessageEvent(eventData.event || 'message', {
              data: eventData.data,
              lastEventId: eventData.id || '',
            });
            this.onmessage(messageEvent);
          }
        }
      }
    }
  } catch (error) {
    // 에러 발생 시, 연결 상태가 CLOSED가 아니라면 onerror 콜백 호출
    if (this.readyState !== 2 && this.onerror) {
      this.onerror(new Event('error'));
    }
    console.error('CustomSSE connection error:', error);
  }
}
 
close() {
  this.readyState = 2; // CLOSED 상태로 변경
  this._abortController.abort(); // fetch 요청을 취소하여 연결을 종료합니다.
}

function EmailCampaignProgress({ campaignId }: { campaignId: string }) {
  const [progress, setProgress] = useState({ sent: 0, failed: 0, total: 0 });
  const [status, setStatus] = useState<'connecting' | 'sending' | 'done' | 'error'>('connecting');
 
  useEffect(() => {
    const sse = new CustomSSE(`/api/campaigns/${campaignId}/progress`);
 
    sse.onopen = () => {
      setStatus('sending');
    };
 
    sse.onmessage = (event) => {
      const data = JSON.parse(event.data);
      setProgress({ sent: data.sent, failed: data.failed, total: data.total });
 
      if (data.sent + data.failed >= data.total) {
        setStatus('done');
        sse.close();
      }
    };
 
    sse.onerror = () => {
      setStatus('error');
    };
 
    sse.connect();
 
    return () => sse.close();
  }, [campaignId]);
 
  return (
    <div>
      <p>{status === 'sending' && `발송 중... ${progress.sent}/${progress.total}`}</p>
      <p>{status === 'done' && `완료 (성공 ${progress.sent}, 실패 ${progress.failed})`}</p>
      <p>{status === 'error' && '연결이 끊어졌습니다.'}</p>
    </div>
  );
}