NestJS SSE (Server-Sent Events)

GET /api/feed/trend/sse HTTP/1.1
Accept: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"message":"현재 트렌드 피드 수신 완료","data":[...]}

data: {"message":"새로운 트렌드 피드 수신 완료","data":[...]}

id: 1
event: ranking-update
data: {"trend": [...]}
retry: 5000

import { Controller, Sse } from '@nestjs/common';
import { Observable } from 'rxjs';

@Controller('feed')
export class FeedController {
  @Sse('trend/sse')
  readTrendFeedList(): Observable<MessageEvent> {
    return new Observable((observer) => {
      // observer.next()를 호출할 때마다 클라이언트에 이벤트 전송
      observer.next({
        data: { message: '초기 데이터', data: [] },
      });
    });
  }
}

interface MessageEvent {
  data: string | object;  // → data: 필드
  id?: string;            // → id: 필드
  type?: string;          // → event: 필드
  retry?: number;         // → retry: 필드
}

import { Controller, Sse } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter';
import { Observable } from 'rxjs';

@Controller('feed')
export class FeedController {
  constructor(
    private readonly feedService: FeedService,
    private readonly eventService: EventEmitter2,
  ) {}

  @Sse('trend/sse')
  readTrendFeedList() {
    return new Observable((observer) => {
      // 1. 연결 즉시 현재 데이터 전송
      this.feedService
        .readTrendFeedList()
        .then((trendData) => {
          observer.next({
            data: {
              message: '현재 트렌드 피드 수신 완료',
              data: trendData,
            },
          });
        })
        .catch((err) => observer.error(err));

      // 2. 이후 데이터 변경 시마다 전송
      const handler = (trendData) => {
        observer.next({
          data: {
            message: '새로운 트렌드 피드 수신 완료',
            data: trendData,
          },
        });
      };

      this.eventService.on('ranking-update', handler);

      // 3. 연결 종료 시 리스너 정리 (중요!)
      return () => {
        this.eventService.off('ranking-update', handler);
      };
    });
  }
}

import { Injectable } from '@nestjs/common';
import { Cron, CronExpression } from '@nestjs/schedule';
import { EventEmitter2 } from '@nestjs/event-emitter';

@Injectable()
export class FeedScheduler {
  constructor(
    private readonly feedService: FeedService,
    private readonly eventService: EventEmitter2,
  ) {}

  @Cron(CronExpression.EVERY_30_SECONDS)
  async updateTrendFeed() {
    const trendData = await this.feedService.calculateTrendRanking();
    // 연결된 모든 SSE 클라이언트에게 전달됨
    this.eventService.emit('ranking-update', trendData);
  }
}

const eventSource = new EventSource('/api/feed/trend/sse');

// 기본 message 이벤트 수신
eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data);
};

// 에러 처리 (네트워크 끊김 등)
eventSource.onerror = (error) => {
  console.error('SSE 연결 에러:', error);
  // 브라우저가 자동으로 재연결을 시도함
};

// 연결 종료
eventSource.close();

// 서버에서 { type: 'ranking-update', data: ... } 형태로 보낸 경우
eventSource.addEventListener('ranking-update', (event) => {
  const data = JSON.parse(event.data);
  handleRankingUpdate(data);
});

eventSource.addEventListener('new-feed', (event) => {
  const data = JSON.parse(event.data);
  handleNewFeed(data);
});

import { useEffect } from 'react';
import { useQueryClient } from '@tanstack/react-query';

export const useTrendingPosts = () => {
  const queryClient = useQueryClient();

  useEffect(() => {
    const eventSource = new EventSource(`${BASE_URL}/feed/trend/sse`);

    eventSource.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        // React Query 캐시에 직접 데이터 주입
        queryClient.setQueryData(['trending-posts'], data);
      } catch (e) {
        console.error('SSE 데이터 파싱 에러:', e);
      }
    };

    return () => {
      eventSource.close();
    };
  }, [queryClient]);
};

@Sse('trend/sse')
readTrendFeedList() {
  return new Observable((observer) => {
    // 30초마다 빈 이벤트 전송 (keep-alive)
    const heartbeat = setInterval(() => {
      observer.next({ data: '' });
    }, 30000);

    // 실제 데이터 전송 로직...

    return () => {
      clearInterval(heartbeat);
    };
  });
}

location /api/feed/trend/sse {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;         # 버퍼링 비활성화 (중요!)
    proxy_cache off;
    proxy_read_timeout 86400s;   # 24시간
}

import { interval, map, switchMap } from 'rxjs';

@Sse('metrics/sse')
streamMetrics() {
  return interval(5000).pipe(
    switchMap(() => this.metricsService.getCurrentMetrics()),
    map((metrics) => ({
      data: { message: '메트릭 갱신', data: metrics },
    })),
  );
}

const response = await fetch('/api/sse', {
  headers: { Authorization: `Bearer ${token}` },
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const text = decoder.decode(value);
  // SSE 형식 파싱...
}