[SNS] Ch04. SSE로 알람 개선

공부 내용을 정리하고 앞으로의 학습에 이해를 돕기 위해 작성합니다.

프론트엔드와 백엔드에서 알림 기능을 실시간으로 제공하기 위해 SSE(Server-Sent Events) 기능을 도입하여 알림 페이지를 개선한다. 기존에는 새로고침을 통해서만 데이터를 업데이트할 수 있었으나, 이번 개선 작업을 통해 실시간으로 알림이 전달되도록 한다.

 

1. 프론트엔드 코드 업데이트 (index.js)

function Alarm() {
  ...
  
  const [alarmEvent, setAlarmEvent] = useState(undefined);
	
  ...
  
  let eventSource = undefined;
...

url: '/api/v1/users/alarm?size=5&sort=id,desc&page=' + pageNum,

...

useEffect(() => {
    handleGetAlarm();

    eventSource = new EventSource("http://localhost:8080/api/v1/users/alarm/subscribe?token=" + localStorage.getItem('token'));

    setAlarmEvent(eventSource);

    eventSource.addEventListener("open", function (event) {
      console.log("connection opened");
    });

    eventSource.addEventListener("alarm", function (event) {
       console.log(event.data);
       handleGetAlarm();
    });

    eventSource.addEventListener("error", function (event) {
      console.log(event.target.readyState);
      if (event.target.readyState === EventSource.CLOSED) {
        console.log("eventsource closed (" + event.target.readyState + ")");
      }
      eventSource.close();
    });

  },

• 수정 전
  • 기본적으로 React의 useState, useEffect 훅을 사용하여 페이지를 구성하고, 알림 데이터를 서버에서 받아와서 렌더링 하는 구조이다.
  • 알림 리스트는 처음 한 번만 API를 호출해 가져오고, 새로고침이나 페이지 이동이 있어야 데이터를 다시 불러온다.
  • handleGetAlarm 함수는 특정 페이지의 알림 데이터를 서버에서 가져와 상태를 업데이트하는 함수로, axios를 사용하여 API 요청을 보낸다.
  • 알림 데이터를 서버에서 받아오면 alarms 배열에 저장하고, 페이지네이션 처리 등을 통해 알림을 리스트 형식으로 출력한다.
• 수정 후
  • EventSource 추가: SSE를 통해 서버와 실시간 연결을 유지한다.
    • useEffect에서 EventSource 객체를 생성해 서버와 연결을 설정한다. 이때, 구독 요청을 보내기 위해 API URL에 JWT 토큰을 쿼리 파라미터로 전달한다.
    • 이벤트가 발생하면("alarm" 이벤트), 서버로부터 데이터를 받아 handleGetAlarm()을 다시 호출해 알림 리스트를 갱신한다.
    • eventSource.addEventListener를 사용해 이벤트 발생 시 동작할 함수를 등록하고, 연결 오류가 발생했을 때는 연결을 종료한다.
  • SSE 연결 이벤트
    • "open": SSE 연결이 성공적으로 열리면 로그를 출력한다.
    • "alarm": 서버로부터 새로운 알람이 발생한 경우, 해당 데이터를 출력하고 다시 handleGetAlarm을 호출해 최신 알림 데이터를 받아온다.
    • "error": 연결이 끊어지거나 오류가 발생하면 해당 이벤트가 실행되어 SSE 연결을 종료한다.

2. JWT 토큰 처리 수정 (JwtTokenFilter)

서버에서 JWT 토큰을 처리하는 부분을 SSE 구독 요청에 맞게 수정한다.

...
	private final static List<String> TOKEN_IN_PARAM_URLS = List.of("/api/v1/users/alarm/subscribe");

    ...

        // get header
        final String token;

        try {
            if (TOKEN_IN_PARAM_URLS.contains(request.getRequestURI())) {
                log.info("Request with {} check the query param", request.getRequestURI());
                token = request.getQueryString().split("=")[1].trim();
            }else{
                final String header = request.getHeader(HttpHeaders.AUTHORIZATION);
                if (header == null || !header.startsWith("Bearer ")) {
                    log.error("Error occurs while getting header. header is null or invalid");
                    filterChain.doFilter(request, response);
                    return;
                }
                token = header.split(" ")[1].trim();
            }

            if (JwtTokenUtils.isExpired(token, key)) {
                log.error("Key is expired");
                filterChain.doFilter(request, response);
                return;
            }
            
            ...

수정 전

• Authorization 헤더에서 JWT 토큰을 추출하여 인증을 처리하는 구조이다. 모든 요청에 대해 JWT가 만료되었는지, 유효한지 검증한 후, 검증이 완료되면 사용자 정보를 보안 컨텍스트에 저장한다.

수정 후

• SSE 구독 요청은 Authorization 헤더가 아닌 쿼리 파라미터로 JWT 토큰을 전달받는다.
  • 구독 요청에 대해 TOKEN_IN_PARAM_URLS 리스트를 참조해 요청 URL이 포함되어 있으면, 쿼리 파라미터에서 JWT 토큰을 추출한다.
  • 일반적인 API 요청은 여전히 Authorization 헤더에서 토큰을 추출해 처리하지만, SSE 구독 요청은 URL 쿼리 파라미터로 JWT 토큰을 처리한다.

3. UserController에 알람 구독 API 추가 

private final AlarmService alarmService;

...


@GetMapping("/alarm/subscribe")
public SseEmitter subscribe(Authentication authentication) {
    User user = ClassUtils.getSafeCastInstance(authentication.getPrincipal(), User.class).orElseThrow(() -> new SnsApplicationException(ErrorCode.INTERNAL_SERVER_ERROR,
            "Casting to User class failed"));
    return alarmService.connectAlarm(user.getId());
}

• @GetMapping("/alarm/subscribe") API 엔드포인트를 통해 SSE 구독 요청을 처리한다.
• 인증된 사용자가 구독 요청을 보내면, SseEmitter 객체를 반환하여 클라이언트와 실시간 연결을 유지한다.
• 사용자 정보는 Authentication 객체에서 가져오고, 그 정보를 통해 알림 구독을 위한 연결을 설정한다.

4. ErrorCode에 새로운 에러 코드 추가

ALARM_CONNECT_ERROR(HttpStatus.INTERNAL_SERVER_ERROR, "Connecting alarm occurs error");

• ALARM_CONNECT_ERROR: SSE 연결 중 발생할 수 있는 오류를 처리하기 위해 새로운 에러 코드를 추가한다. 알람 구독 요청 처리 중 예외가 발생하면 이 에러 코드로 예외를 처리할 수 있다.

5. EmitterRepository 추가

package com.example.sns.repository;

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 {

    private Map<String, SseEmitter> emitterMap = new HashMap<>();

    public SseEmitter save(Integer userId, SseEmitter sseEmitter) {
        final String key = getKey(userId);
        emitterMap.put(key, sseEmitter);
        log.info("Set sseEmitter {}", userId);
        return sseEmitter;
    }

    public Optional<SseEmitter> get(Integer userId) {
        final String key = getKey(userId);
        log.info("Get sseEmitter {}", userId);
        return Optional.ofNullable(emitterMap.get(key));
    }

    public void delete(Integer userId) {
        emitterMap.remove(getKey(userId));
    }

    private String getKey(Integer userId) {
        return "Emitter:UID: " + userId;
    }
}

• 사용자의 SseEmitter 객체를 저장하고 관리하는 역할을 한다.
• 저장(save), 조회(get), **삭제(delete)**와 같은 메서드를 제공하며, 각 사용자의 구독 정보를 관리한다.
• emitterMap을 통해 userId별로 SseEmitter 객체를 저장하여 실시간 알림을 전송할 때 사용할 수 있다.

6. AlarmService 추가

package com.example.sns.service;

import com.example.sns.exception.ErrorCode;
import com.example.sns.exception.SnsApplicationException;
import com.example.sns.repository.EmitterRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;

@Slf4j
@Service
@RequiredArgsConstructor
public class AlarmService {

    private final static Long DEFAULT_TIMEOUT = 60L * 1000 * 60;
    private final static String ALARM_NAME = "alarm";
    private final EmitterRepository emitterRepository;

    public void send(Integer alarmId, Integer userId) {
        emitterRepository.get(userId).ifPresentOrElse(sseEmitter -> {
            try {
                sseEmitter.send(SseEmitter.event().id(alarmId.toString()).name(ALARM_NAME).data("new alarm"));
            } catch (IOException e) {
                emitterRepository.delete(userId);
                throw new SnsApplicationException(ErrorCode.ALARM_CONNECT_ERROR);
            }
        }, () -> log.info("No emitter founded"));
    }

    public SseEmitter connectAlarm(Integer userId) {
        SseEmitter sseEmitter = new SseEmitter(DEFAULT_TIMEOUT);
        emitterRepository.save(userId, sseEmitter);
        sseEmitter.onCompletion(() -> emitterRepository.delete(userId));
        sseEmitter.onTimeout(() -> emitterRepository.delete(userId));

        try {
            sseEmitter.send(SseEmitter.event().id("").name(ALARM_NAME).data("connect completed"));
        } catch (IOException e) {
            throw new SnsApplicationException(ErrorCode.ALARM_CONNECT_ERROR);
        }

        return sseEmitter;
    }
}

• SseEmitter를 관리하고, 알림이 발생했을 때 해당 사용자에게 실시간으로 알림을 전송하는 역할을 한다.
• connectAlarm: 사용자가 알림 구독을 요청하면 새로운 SseEmitter 객체를 생성하고, 이를 관리하도록 EmitterRepository에 저장한다. 연결이 종료되거나 타임아웃되면 emitter를 삭제한다.
• send: 알림이 발생하면 해당 사용자에게 SseEmitter를 통해 실시간으로 알림 데이터를 전송한다.

7. PostService 수정 (PostService.java)

private final AlarmService alarmService;
...

@Transactional
public void like(Integer postId, String userName) {
   ...

    // like save
    likeEntityRepository.save(LikeEntity.of(userEntity, postEntity));
    AlarmEntity alarmEntity = alarmEntityRepository.save(AlarmEntity.of(postEntity.getUser(), AlarmType.NEW_LIKE_ON_POST, new AlarmArgs(userEntity.getId(), postEntity.getId())));

    alarmService.send(alarmEntity.getId(), postEntity.getUser().getId());
}

@Transactional
public void comment(Integer postId, String userName, String comment) {
    ...

    // comment save
    commentEntityRepository.save(CommentEntity.of(userEntity, postEntity, comment));
    AlarmEntity alarmEntity = alarmEntityRepository.save(AlarmEntity.of(postEntity.getUser(), AlarmType.NEW_COMMENT_ON_POST, new AlarmArgs(userEntity.getId(), postEntity.getId())));

    alarmService.send(alarmEntity.getId(), postEntity.getUser().getId());

}

...

• like, comment 메서드에서 게시물에 대한 알림이 발생할 때 알림 데이터를 저장하고, 알림을 구독 중인 사용자에게 실시간으로 알림을 전송한다.
• alarmService.send() 메서드를 호출하여 새로운 좋아요나 댓글이 달릴 때 실시간 알림을 전송한다.

실행 테스트

admin1 계정으로 웹페이지에 로그인하고, admin 계정으로 Postman에서 특정 게시물에 댓글을 달아 실시간으로 화면에 반영되는지 확인한다.

 

1. admin1 로그인: 웹페이지에서 admin1 계정으로 로그인하여 MyPosts 탭에서 자신이 작성한 게시물들을 확인한다.

 

2. Postman을 통한 댓글 작성

• Postman을 사용해 admin 계정으로 10번 게시글에 댓글을 작성한다.
• POST /api/v1/posts/10/comments 엔드포인트에 댓글을 전송하면 서버에서 실시간으로 처리되어 admin1의 화면에 댓글이 바로 반영된다.
• SSE(Server-Sent Events)를 사용한 실시간 알림 및 업데이트가 정상적으로 동작하는 것을 확인할 수 있다.

이번 테스트를 통해 SSE를 활용한 실시간 알림 기능이 정상적으로 동작함을 확인했다. 댓글이나 좋아요와 같은 이벤트가 발생할 때마다 페이지가 실시간으로 갱신되며, 별도의 새로고침 없이도 최신 상태를 유지할 수 있다.