[웹] Server Sent Event

포트폴리오를 만들던 도중 다음과 같은 상황을 만나게 되었다.

 

사용자가 동영상을 업로드 한다.

업로드된 동영상은 플랫폼이 제공하는 기능들을 위한 후처리 작업을 수행해야 한다. (포맷 통일, 자막 생성, 썸네일 생성, 메타데이터 추출)

사용자의 동영상이 후처리가 완료된 이후, 서버에선 사용자에게 업로드가 성공되었다는 알람을 보내줘야 한다.

 

위의 작업들 중 특히 자막 생성은 Whisper AI를 사용해서 구현되어 있는데, 내 개발 PC에선 단일 작업만 허용된다. (vRAM 제한)

또한 15분짜리 동영상 기준 2-3분 정도의 작업 시간을 필요로 한다.

일단 사용자가 업로드 로직으로 오랜 시간 쓰레드를 붙잡고 있게 하면 안되기 때문에 업로드 로직과 후처리 로직을 분리했다.

 

사용자의 파일 업로드 -> 서버의 파일 수신 및 저장 -> 파일 저장 완료 응답 생성 (쓰레드 반환) 및 후처리 이벤트 생성(MQ)

후처리 쓰레드에서 후처리 작업 실행(병렬처리 없이 단일 쓰레드로 동작) -> 처리 완료 후, 사용자에게 성공 알림 제공

 

파일을 업로드하는 로직도 사실 상당한 시간을 소요하기 때문에 분리하고 싶긴 하지만......

어차피 포트폴리오다. 주석만 남겨두고...... 패스

다만 별도의 서버로 업로드 서버를 만들기 위한 절차적 로직만 나중에 개발해두어야겠다.

 

 

 

여기서 일단 알림 로직을 구현하기 위해 기술 검토를 수행했다.

 

알림 기능을 구현하기 위해서는 주로 3가지의 방법을 사용할 것이다.

1. 주기적인 서버 조회

- React-Query나 setInterval 기능을 이용해서 주기적으로 서버에 질의하고, 응답을 받아서 사용자에게 알리는 구조이다.

가장 쉽게 떠올릴 수 있는 방법이고, 실제로 구현도 쉽다. 나도 첫 프로젝트에서 모니터링해야하는 정보들이 있어서 이 방식으로 구현했었다.

사용자 수가 많지 않고, DB에 큰 부하가 가해지지 않는 서비스라면 고려해볼 수 있다.

클라이언트의 수가 많아질수록 서버의 부하 역시 정비례로 늘어난다는 점을 유념해야한다.

DB 부하를 줄이기 위해 서버에서 짧은 시간으로 Caching을 도입하는 것도 좋은 방법이 될 수 있겠다.

 

2. SSE(Server Sent Event)

- 내가 검토중인 SSE 기능이다. 최초에 클라이언트가 서버에 HTTP 연결을 보내고, 서버는 해당 요청에 필요한 Response를 저장한다.

만약 서버에서 클라이언트에 메시지를 보내야 하는 상태가 되면, 저장된 해당 클라이언트의 Response를 찾아 응답 메시지를 보낸다.

서버는 클라이언트에게 Response를 기억할 시간을 지정해야 하며, 해당 시간이 지나기 전 클라이언트와 통신하여 시간을 연장한다.

클라이언트는 서버에 저장한 요청으로부터 응답 신호를 1회 이상 수신해야 한다. 이를 위해 open 메시지를 수신하는 것이 권장된다.

즉, 클라이언트와 서버의 연결 이후 일정 시간 동안 유지되는 Response를 서버에서 보관하고 있다가, 필요할 때 사용한다. 그래서 서버에서 클라이언트로 향하는 단방향 메시지에 적합하다.

 

3. WebSocket

- 웹 표준 프로토콜 방식 중 하나로 서버와 클라이언트가 자주 메시지를 주고받는 상황에서 사용하기 좋다. 연결된 서버와 클라이언트 간에는 거의 실시간으로 메시지를 교환할 수 있다는 장점이 있다. 양방향으로 메시지를 주고받기 때문에 채팅 등의 기능을 만들 때 많이 사용된다. 웹소켓은 연결을 위해 HTTP를 사용한다. 이를 위해 연결 시, 프로토콜 전환을 위한 메시지를 주고받고, 연결이 완료되면 웹소켓 프로토콜로 전환되어 양방향 통신이 수행된다.

 

 

---

1. 프론트엔드

SSE 자체는 기본 스펙이다보니, 웹에서 일반적인 방식으로도 충분히 사용할 수 있다.

다만, Header 등 섬세한 조율이 필요한 경우 event-source-polyfill과 같은 라이브러리의 도움을 받는 것이 좋다.

XMLHttpRequest로 서버에 요청하고 스트림 데이터를 받아서 처리하는 방식으로 내부 구현이 되어 있다고 한다.

학원에서 jQuery를 배우면서 직접 ajax 요청을 날린 적이 없었는데, 오랜만에 만난 객체...

 

  useEffect(() => {
    if (eventSourceRef.current) return;
    const eventSourcePolyfillInit: EventSourcePolyfillInit = {
      headers: {
        Authorization: `Bearer ${options?.token}`,
      },
      heartbeatTimeout: 60000,
    };
    const eventSource = new EventSourcePolyfill(subscribeUrl, eventSourcePolyfillInit);

    eventSource.addEventListener('connection', () => {
      setIsConnected(true);
      if (options && options.onOpen) options.onOpen();
    });

    eventSource.addEventListener('upload_notification', (event: MessageEvent) => {
      console.log(event.data);
      setMessages((prev) => [...prev, event.data]);
    });

    eventSource.onerror = (error: any) => {
      console.error('EventSource failed:', error);
      setIsConnected(false);
      eventSource.close();
      if (options && options.onError) options.onError(error);
    };

    eventSourceRef.current = eventSource;

    return () => {
      if (eventSourceRef.current) {
        setIsConnected(false);
        eventSource.close();
        eventSourceRef.current = undefined;
      }
    };
  }, [options, options?.token, subscribeUrl]);

 

여기서 꽤 오랜 시간을 소요했는데, 나같은 실수를 한 글이 많더라.

    @GetMapping(path = "/subscribe/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribeToVideo(@PathVariable(name = "userId") Long userId) {
        SseEmitter emitter = new SseEmitter(SSE_CONNECTION_TIMEOUT);
        emitters.put(userId, emitter);

        // 타임아웃 또는 완료 시 리스트에서 emitter 제거
        emitter.onTimeout(() -> {
            log.info("SSE Emitter 시간 제한: {}", userId);
            emitters.remove(userId);
        });
        emitter.onCompletion(() -> {
            log.info("SSE Emitter 완료: {}", userId);
            emitters.remove(userId);
        });
        emitter.onError(e -> {
            log.error("SSE Emitter 에러: {}: {}", userId, e.getMessage());
            emitters.remove(userId);
        });

        try {
            emitter.send(SseEmitter.event()
                    .id(String.valueOf(System.currentTimeMillis()))
                    .name("connection") // 이벤트 이름
                    .data("SSE 연결 성공 " + userId)
                    .reconnectTime(10000)); // 재연결 시간 (ms)
        } catch (IOException e) {
            log.error("ERROR {}: {}", userId, e.getMessage());
            emitter.completeWithError(e);
            emitters.remove(userId);
        }

        return emitter;
    }

 

아래에서 나는 분명히 연결 직후, .name("connection") 이라는 이름으로 최초 메시지를 프론트에 제공했다.

이 메시지는 위의 프론트 코드에서 addEventListener("connection", () => void);에 대응하게 된다.

그런데 그냥 .onopen() 함수에서 왜 연결이 성공했는데 메시지가 안나오지? 라며 고민했다.

마찬가지로 메시지가 .onmessage() 함수에서 받아지지 않는다고 고민했다.

 

서버에서 보내는 .name에 들어 있는 이름으로 메시지가 전송되고, 화면에서는 해당 이름으로 이벤트 리스너를 만들어야 한다.

이 부분을 몰라서 거의 하루를 잡아먹었다.

 

1. EventSource 객체는 일정시간 heartBeat가 도착하지 않으면 연결이 실패했다고 생각하고 연결을 끊어버린다.(default 45초)

2. EventSource 객체는 서버에서 전송한 스트림 데이터를 name으로 잘라서 이벤트를 발행한다.

 

따라서 open이 성공적으로 이뤄졌지만, name으로 수신한 데이터가 하나도 없을 경우 연결을 끊어버린다.

이 부분을 몰라서 onmessage만 죽어라 살펴보고 있었다.

 

---

하여튼 알림 부분도 pub/sub 구조로 확장해 나가야하는 부분인 것 같다.

나중에 공지사항 등에 대한 알림을 전체 사용자에게 보내거나 하게 되면 부하가 발생할 포인트가 되기 좋다.

그리고 다중화된 서버에서 알림을 보내는 경우에 대해서도 생각해보아야겠다.

서버 A에 emitter가 저장되어 있는데, 서버 B에서 알림에 대한 메시지를 소비해 버리는 경우 어떻게 될까?

 

역시... 기능은 만들면 만들수록 더 많은 고민을 필요로 한다.