알림 기능 구현하기

참고자료
https://taemham.github.io/posts/Implementing_Notification/
https://tecoble.techcourse.co.kr/post/2022-10-11-server-sent-events/

NotificationController

package com.site.xidong.notification;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RequiredArgsConstructor
@RestController
@RequestMapping("/notification")
@Slf4j
public class NotificationController {
    private final NotificationService notificationService;

    @GetMapping("/subscribe")
    public SseEmitter subscribe() throws Exception {
        return notificationService.connectNotification();
    }
}

NotificationService

package com.site.xidong.notification;

import com.site.xidong.security.SiteUserSecurityDTO;
import com.site.xidong.siteUser.SiteUser;
import com.site.xidong.siteUser.SiteUserRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;

@Service
@RequiredArgsConstructor
@Slf4j
public class NotificationService {
    private final static Long DEFAULT_TIMEOUT = 60 * 60 * 1000L; // 1시간
    private final static String CONNECTION = "connection";
    private final static String NEW_COMMENT = "new_comment";

    private final SiteUserRepository siteUserRepository;
    private final EmitterRepository emitterRepository;

    public SseEmitter connectNotification() throws Exception {
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        SiteUserSecurityDTO siteUserSecurityDTO = (SiteUserSecurityDTO) auth.getPrincipal();
        SiteUser siteUser = siteUserRepository.findSiteUserByUsername(siteUserSecurityDTO.getUsername()).get();
        String username = siteUser.getUsername();

        // 새로운 SseEmitter를 만든다
        SseEmitter sseEmitter = new SseEmitter(DEFAULT_TIMEOUT);

        // username으로 SseEmitter를 저장
        emitterRepository.save(username, sseEmitter);

        // 세션이 종료될 경우 저장한 SseEmitter를 삭제한다. (로그아웃되면 다시 클라이언트에서 연결 요청 보내라는 건가?)
        sseEmitter.onCompletion(() -> emitterRepository.delete(username));
        sseEmitter.onTimeout(() -> emitterRepository.delete(username));

        // 503 Service Unavailable 오류가 발생하지 않도록 첫 데이터를 보낸다.
        try {
            sseEmitter.send(SseEmitter.event().id("connection-established-001").name(CONNECTION).data("Connection completed!"));
        } catch (IOException exception) {
            throw new Exception("Failed to Connect SSE");
        }
        return sseEmitter;
    }

    public void send(String username, Long notificationId) {
        // username으로 SseEmitter를 찾아 이벤트를 발생 시킨다.
        emitterRepository.get(username).ifPresentOrElse(sseEmitter -> {
            try {
                sseEmitter.send(SseEmitter.event().id(notificationId.toString()).name(NEW_COMMENT).data("New comment!"));
            } catch (IOException exception) {
                // IOException이 발생하면 저장된 SseEmitter를 삭제하고 예외를 발생시킨다.
                emitterRepository.delete(username);
                throw new Error("Failed to Connect SSE");
            }
        }, () -> log.info("No Emitter Found"));
    }
}

EmitterRepository

package com.site.xidong.notification;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Repository;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

@Slf4j
@Repository
public class EmitterRepository {
    // username을 키로 SseEmitter를 해시맵에 저장
    private Map<String, SseEmitter> emitterMap = new HashMap<>();

    public SseEmitter save(String username, SseEmitter sseEmitter) {
        emitterMap.put(getKey(username), sseEmitter);
        log.info("Saved SseEmitter for {}", username);
        return sseEmitter;
    }

    public Optional<SseEmitter> get(String username) {
        log.info("Got SseEmitter for {}", username);
        return Optional.ofNullable(emitterMap.get(getKey(username)));
    }

    public void delete(String username) {
        emitterMap.remove(getKey(username));
        log.info("Deleted SseEmitter for {}", username);
    }

    private String getKey(String username) {
        return "Emitter:username:" + username;
    }
}

/notification/subscribe API 테스트를 위해 인스턴스에 배포 후 Swagger로 테스트를 진행한다.

403 에러가 발생했다. /notification/subscribe uri를 인증이 필요한 uri 목록에 추가한 후 다시 테스트 해 보자.

무한 로딩이 발생했다.
로그를 추가한 뒤 다시 테스트 해 보자.

연결이 정상적으로 된 것 같다.

NotificationController를 수정해준다.

@GetMapping("/subscribe")
    public ResponseEntity<SseEmitter> subscribe() throws Exception {
        SseEmitter sseEmitter;
        sseEmitter = notificationService.connectNotification();
        return ResponseEntity.status(HttpStatus.OK).body(sseEmitter);
    }

createComment 메소드에 send()를 호출하는 로직을 추가해보자.

CommentService

import com.site.xidong.security.SiteUserSecurityDTO;
import com.site.xidong.siteUser.SiteUser;
import com.site.xidong.siteUser.SiteUserRepository;
import com.site.xidong.video.Video;
import com.site.xidong.video.VideoRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

@Log4j2
@Service
@RequiredArgsConstructor
public class CommentService {
    private final CommentRepository commentRepository;
    private final SiteUserRepository siteUserRepository;
    private final VideoRepository videoRepository;

    public CommentReturnDTO create(Long videoId, String contents) {
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        SiteUserSecurityDTO siteUserSecurityDTO = (SiteUserSecurityDTO) auth.getPrincipal();
        SiteUser siteUser = siteUserRepository.findSiteUserByUsername(siteUserSecurityDTO.getUsername()).get();
        Video video = videoRepository.findById(videoId).get();
        Comment comment = Comment.builder()
                .siteUser(siteUser)
                .video(video)
                .contents(contents)
                .createdAt(LocalDateTime.now())
                .build();
        Comment newComment = commentRepository.save(comment);
        
        // 알림 보내기
        notificationService.send(video.getSiteUser().getUsername(), newComment.getContents());
        
        CommentReturnDTO commentReturnDTO = CommentReturnDTO.builder()
                .commentId(newComment.getId())
                .imageUrl(newComment.getSiteUser().getImageUrl())
                .username(newComment.getSiteUser().getUsername())
                .nickname(newComment.getSiteUser().getNickname())
                .videoPath(newComment.getVideo().getVideoPath())
                .videoName(newComment.getVideo().getVideoName())
                .createdAt(newComment.getCreatedAt())
                .updatedAt(null)
                .contents(newComment.getContents())
                .build();
        return commentReturnDTO;
    }
    /* 생략 */
}

포스트맨으로 /notification/subscribe에 get 요청을 테스트 한 결과 SSE 연결이 완료 되었지만 sseEmitter가 리턴값으로 오지 않았다.

왜 그런걸까?

🔥 핵심 원인: Postman은 SSE를 지원하지 않음
Postman은 Server-Sent Events (SSE) 와 같이 지속적인 스트리밍 응답을 처리할 수 없다.

💡 왜 그런가?
SSE는 서버가 클라이언트로 지속적으로 데이터를 보내는 단방향 스트리밍 프로토콜인데,
Postman은 요청-응답 구조에 최적화되어 있어서, 이런 열려 있는 연결(streaming response) 을 끝나지 않은 것으로 인식하고 결과를 안 보여준다.

Postman에서는 SSE 연결 테스트가 불가능하다.
브라우저에서 EventSource로 테스트하거나, curl로 테스트해야 한다.

curl로 테스트 해보자.

curl -H "Authorization: Bearer {토큰 값}" https://{서버 ip 주소}.nip.io/notification/subscribe

사용자의 영상에 새로운 댓글이 달렸을 때 알림이 오는지도 테스트 해 보자.

아무런 알림이 오지 않는다.

EmitterRepository를 싱글톤으로 설정해주지 않아서 생긴 문제일 수 있다.

EmitterRepository 클래스에 Spring의 어노테이션(@Component)을 추가하지 않았다.

즉, 현재 상태로는 Spring 빈이 아니고,
@RequiredArgsConstructor를 통해 새로운 EmitterRepository 객체를 생성하고 있다.

이럴 경우, NotificationService가 요청마다 새로운 EmitterRepository 인스턴스를 가지게 된다.
→ 그래서 subscribe할 때 저장된 emitter는 send할 때 못 찾게 된다.

SseEmitter는 사용자마다 1개씩 서버에 보관되어야 하고,
이걸 저장하는 Map (emitterMap)은 모든 요청에서 공유되는 전역 저장소여야 한다.

즉:

A 사용자가 /subscribe로 접속 → emitterMap.put("A", emitter)

이후 A에게 send("A", message) 호출 → emitterMap.get("A") → 전송

이 흐름이 되려면 emitterMap을 담고 있는 EmitterRepository는 하나만 존재해야 함 = 싱글톤

EmitterRepository.java에 @Component 어노테이션을 붙여준 후 다시 테스트 해 보자.

알림이 잘 오는 것을 확인할 수 있다.