[Nest.js] SSE를 이용하여 알림 기능 구현

1. 개요

이번 프로젝트에서 알림기능을 도입하게 되었는데, 먼저 떠오른건 소켓통신이다.

알림기능이란 것은 결국 이벤트가 발생했을때 클라이언트로 부터 별도의 요청없이 서버에서 클라이언트로 메세지를 보내야 함을 의미한다. 이전에 채팅방을 구현해야 했을때 소켓통신을 다룬적이 있는데 이가 적합해보였다.

하지만, 알림 기능은 채팅보다는 비교적 간단한 기능이고, 서버에서 클라이언트로 메세지를 보내는 다른 방법이 없을까 해서 찾아본 결과 SSE에 대해서 알게되었다.

2. Socket vs SSE

먼저 알림기능 구현에 앞서 어떤 기술을 사용할지 결정하기 전에 더 적합한 것을 고르기 위해 각 기술의 장단점을 정리해보았다.

Socket의 장단점 및 특징

• 양방향 통신
• 낮은 지연 시간 (연결 후 추가적인 헤더 교환이 없으므로 효율적)
• 데이터를 자주 주고 받는 상황이 아닌 경우, 오히려 리소스 소모가 큼.

SSE의 장단점 및 특징

• 단방향 통신
• 간단한 구현 (소켓에 비해 구현이 간단)
• 데이터 전송 제한 (텍스트 데이터만을 전송할 수 있음)
• 클라이언트의 연결이 끊겨도 알 수 없음

위의 장단점 및 특징들을 정리하고 알림기능 측면에서 다시 생각해보았다. 애초에 소켓통신 외에 다른 방법을 찾아본 이유는 다음과 같다.

1. 구현이 복잡하다.
2. 클라이언트에서 서버로의 데이터 전송은 필요하지 않기에 비효율적

이에 비해 SSE의 단점들은 알림 기능에는 단점이 아니게 되었다. 데이터 전송 제한 또한 어차피 알림에는 텍스트만이 포함되어 있기 때문에 상관없으며, 클라이언트의 연결이 끊겨도 알림을 따로 저장시켜두면 되기에 큰 단점이 아니다.

위와 같은 이유때문에 결국 SSE로 알림기능을 구현하게 되었다.

3. SSE 알림 기능 구현

먼저 알림을 담당하는 모듈을 따로 생성했다.

구현해야 하는 기능은 카드의 담당자가 변경되었을때, 해당 카드의 담당자에게 알림을 보내야한다. 전체적인 알림의 로직은 다음과 같다.

1. 클라이언트가 서버가 SSE 연결 요청을 보낸다.
2. 클라이언트가 SSE 이벤트(Observable)를 구독한다.
3. 카드의 담당자가 업데이트 되는 시점에, 이벤트를 푸쉬한다.
4. SSE 이벤트(Observable)에서 유저 id를 필터링하고 담당자에게만 알림을 보낸다.

sse.controller.ts

import { Controller, Param, Sse } from "@nestjs/common";
import { SseService } from "./sse.service";

@Controller("sse")
export class SseController {
    constructor(private readonly sseService: SseService) {}

    @Sse(":userId")
    sendClientAlarm(@Param("userId") userId: string) {
        return this.sseService.sendClientAlarm(+userId);
    }
}

sse.service.ts

import { Injectable, MessageEvent } from "@nestjs/common";
import { Observable, Subject, filter, map } from "rxjs";

@Injectable()
export class SseService {
    private users$: Subject<any> = new Subject();

    private observer = this.users$.asObservable();

  	// 이벤트 발생 함수
    emitCardChangeEvent(userId: number) {
      	// next를 통해 이벤트를 생성
        this.users$.next({ id: userId });
    }
  
  	// 이벤트 연결
    sendClientAlarm(userId: number): Observable<any> {
      	// 이벤트 발생시 처리 로직
        return this.observer.pipe(
          	// 유저 필터링
            filter((user) => user.id === userId),
          	// 데이터 전송
            map((user) => {
                return {
                    data: {
                        message: "카드의 담당자가 사용자님으로 변경되었습니다.",
                    },
                } as MessageEvent;
            }),
        );
    }
}

card.service.ts

...

async changeUser(id: number, { userId }: ChangeUserCardDto) {
    await this.findCardById(id);

    const user = await this.userSerivce.findUserById(userId);
    if (!user) {
        throw new NotFoundException("존재하지 않는 사용자입니다.");
    }

    await this.cardRepository.update(
      	{ id },
        { user },
    );

  	// 카드 업데이트시 이벤트 발생
    this.sseService.emitCardChangeEvent(user.id);

    return {
        message: `${id}번 카드의 담당자를 ${userId}번 사용자로 변경했습니다.`,
    };
}

...

front.js

const userId = "2"; // 유저 아이디

const eventSource = new EventSource(`http://localhost:5001/sse/${userId}`);

// SSE 이벤트 수신
eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log("Received data:", data);
};

// SSE 연결이 열렸을 때
eventSource.onopen = () => {
    console.log("SSE connection opened");
};

// SSE 연결이 닫혔을 때
eventSource.onclose = () => {
    console.log("SSE connection closed");
};

// SSE 연결 에러가 발생했을 때
eventSource.onerror = (error) => {
    console.error("SSE connection error:", error);
};

참고

nestjs sse 공식문서
rxjs 공식문서
sse 정리 블로그