SSE를 활용한 실시간 댓글,좋아요 알림기능 구현

목차

1.SSE?

2.프로젝트에 적용

 

1.SSE?

SSE 는 Server Sent Events 로 서버에서 클라이언트로 실시간 데이터를 비동기적으로 전송하는 웹 애플리케이션 통신 기술 중 하나입니다. SSE의 특징으로는 다음 몇가지로 볼 수 있습니다.

1. 단방향 통신: 클라이언트는 서버에게 요청을 보내고, 서버는 클라이언트에게 데이터를 주기적으로 보냅니다. 이는 HTTP 기반의 통신 프로토콜을 사용하며, 클라이언트에서 서버로의 요청만 있고, 서버에서 클라이언트로의 응답만 있습니다.
2. 폴링 대신 이벤트 기반: 일반적인 웹 애플리케이션에서는 폴링(Polling)이나 웹소켓(WebSockets)과 같은 방법을 사용하여 실시간 데이터를 처리합니다. 하지만 SSE는 클라이언트가 서버에게 한 번 연결되면, 서버는 필요할 때마다 데이터를 보내는 이벤트 기반의 방식을 채택합니다.
3. 간단한 프로토콜: SSE는 간단한 텍스트 기반의 프로토콜을 사용합니다. **이벤트 스트림(Event Stream)**이라고 불리는 텍스트 데이터를 통해 서버로부터 클라이언트로 이벤트를 전송합니다.
4. 자동 재연결: 만약 연결이 끊기면, SSE는 자동으로 재연결을 시도합니다. 이는 실시간 통신을 유지하는 데 도움이 됩니다.

아래는 SSE의 구조입니다. 

 

SSE의 기본적인 흐름은 클라이언트가 SSE요청을 보내면 서버에서는 클라이언트와 매핑되는 SSE 통신객체를 만들고(SseEmitter) 해당객체가 이벤트 발생시 eventsource를 client에게 전송하면서 데이터가 전달되는 방식입니다. sseemitter는 SSE 통신을 지원하는 스프링에서 지원하는 API입니다.

 

2.프로젝트 적용

지금 진행하고 있는 프로젝트는 게시글에 댓글과 좋아요를 작성을 하면 알림을 보내기만 하면 되는 단순한 구조이기 때문에 WebSocket을 사용할 필요가 없고 SSE를 적용을 하기로 했습니다. 알림기능을 적용을 하기 위해서 다음과 같이 코드를 작성을 했다.

//Controller
@Log4j2
@RestController
@RequestMapping("/api/notice")
@AllArgsConstructor
public class NoticeController {
    private final SSeService service;

    //알림(구독)
    @GetMapping(value = "/subscribe",produces = MediaType.ALL_VALUE)
    public SseEmitter subscribe( @RequestParam(value = "userName") String userName){
        return service.subscribe(userName);
    }
    
    //알림 목록
    @GetMapping("/list/{username}")
    public ResponseEntity<?>noticeList(@PathVariable("username")String username){
        List<NoticeDto>noticeDtoList = service.noticeList(username);
        return new ResponseEntity<>(noticeDtoList,HttpStatus.OK);
    }

}

 

컨트롤러의 타입은 SseEmitter이고 회원의 아이디를 받아서 구독을 하도록 했습니다.

//서비스
@Log4j2
@Service
@AllArgsConstructor
public class SSeService {
    private final EmitterRepositoryImpl emitterRepository;
    private final NotificationRepository notificationRepository;
    private static final Long DEFAULT_TIMEOUT = 60 * 1000L;

    /**
     * 구독 알림
     * @param userName : 회원 아이디
     **/
    public SseEmitter subscribe(String userName){
        String emitterId = makeTimeIncludeId(userName);
        log.info(emitterId);
        SseEmitter emitter = emitterRepository.save(emitterId, new SseEmitter(DEFAULT_TIMEOUT));
        log.info(emitter);
        emitter.onCompletion(() -> emitterRepository.deleteById(emitterId));
        emitter.onTimeout(() -> emitterRepository.deleteById(emitterId));

        // 503 에러를 방지하기 위한 더미 이벤트 전송
        String eventId = makeTimeIncludeId(userName);
        sendNotification(emitter, eventId, emitterId, "EventStream Created. [userId=" + userName + "]");

        return emitter;
    }


    private String makeTimeIncludeId(String memberId) {
        return memberId + "_" + System.currentTimeMillis();
    }


    private void sendNotification(SseEmitter emitter, String eventId, String emitterId, Object data) {
        try {
            emitter.send(SseEmitter
                    .event()
                    .id(eventId)
                    .name("open")
                    .data(data));
        } catch (IOException exception) {
            emitterRepository.deleteById(emitterId);
        }
    }

    private boolean hasLostData(String lastEventId) {
        return !lastEventId.isEmpty();
    }

    private void sendLostData(String lastEventId, Long memberId, String emitterId, SseEmitter emitter) {
        Map<String, Object> eventCaches = emitterRepository.findAllEventCacheStartWithByMemberId(String.valueOf(memberId));
        eventCaches.entrySet().stream()
                .filter(entry -> lastEventId.compareTo(entry.getKey()) < 0)
                .forEach(entry -> sendNotification(emitter, entry.getKey(), emitterId, entry.getValue()));
    }
    
    //알림전송
    public void send(Member receiver, NoticeType notificationType, String content, String data) {

        Notification notification = notificationRepository.save(createNotification(receiver, notificationType, content, data));

        String receiverId = String.valueOf(receiver.getUsername());

        log.info(receiverId);

        String eventId = receiverId + "_" + System.currentTimeMillis();

        log.info(eventId);

        Map<String, SseEmitter> emitters = emitterRepository.findAllEmitterStartWithByMemberId(receiverId);

        log.info(emitters);

        emitters.forEach(
                (key, object) -> {
                    emitterRepository.saveEventCache(key, notification);
                    sendNotification(object, receiverId, key, NoticeDto.create(notification));
                }
        );
    }
    
    //알림 목록
    @Transactional(readOnly = true)
    public List<NoticeDto>noticeList(String username){
        return notificationRepository.findAllByNotification(username);
    }
    
    //알림 개별 조회
    @Transactional(readOnly = true)
    public NoticeDto getNotice(String username,Integer noticeId){
        Optional<NoticeDto>detail = notificationRepository.findByUsername(username,noticeId);
        //조회시 읽은여부 true로 전환하기.
        if(detail.isPresent()){
            Notification notification = Notification
                    .builder()
                    .noticeType(detail.get().getNoticeType())
                    .data(detail.get().getData())
                    .message(detail.get().getMesseage())
                    .noticeType(detail.get().getNoticeType())
                    .isRead(true)
                    .build();
            notificationRepository.save(notification);
        }
        return detail.orElseThrow(()->new CustomExceptionHandler(ErrorCode.NOTICE_NOT_FOUND));
    }
    
    private Notification createNotification(Member member,NoticeType noticeType,String message,String data){
        return Notification
                .builder()
                .member(member)
                .message(message)
                .noticeType(noticeType)
                .isRead(false)
                .data(data)
                .build();
    }
}

 

/**
	  * 댓글 추가하기.
	  * @param dto : 댓글 요청 Dto
	  * @param boardId : 게시글 번호
	  * @param principal : 회원 객체
	  * @exception CustomExceptionHandler : 댓글사용시 로그인을 하지 않은 경우 ONLY_USER
	  * @exception CustomExceptionHandler : 게시판글 조회시 글이 없는 경우에는 NOT_BOARD_DETAIL
	 **/
	@Transactional
	public Integer replyCreate(CommentDto.CommentRequestDto dto,Member principal,Integer boardId){
		
		//유저가 아니면 사용불가
		if(principal == null) {
			throw new CustomExceptionHandler(ErrorCode.ONLY_USER);
		}
		
		//게시판에서 글 조회 -> 글이 없으면 Exception
		Board board = boardrepository.findById(boardId).orElseThrow(()-> new CustomExceptionHandler(ErrorCode.NOT_BOARD_DETAIL));

		Comment reply = Comment.builder()
				.board(board)
				.member(principal)
				.replyWriter(principal.getUsername())
				.replyContents(dto.getReplyContents())
				.createdAt(dto.getCreatedAt())
				.build();

		repository.save(reply);

		board.getCommentlist().add(reply);
		Member writer = board.getWriter();
		//댓글 알림
		sSeService.send(writer, NoticeType.REPLY,"게시글에 댓글이 달렸습니다.",String.valueOf(board.getId()));

		return reply.getId();
	}

/**
 * 좋아요+1
 * @param board : 게시글 객체
 * 게시글조회에서 좋아요 +1기능
**/
@Transactional
public String createLikeBoard(Board board){
    Member member =getMember();
    //좋아요 증가
    board.increaseLikeCount();
    Like like = new Like(board,member);
    repository.save(like);

    //알림기능
    Member writer = board.getWriter();
    sseService.send(writer, NoticeType.LIKE,"게시글에 좋아요가 달렸습니다.", String.valueOf(board.getId()));
    return "좋아요 처리 완료";
}

 

각 서비스에 알림기능을 추가한다.

 

 

코드를 작성했으니 결과를 포스트맨을 활용해서 보도록 하자.

 

 

회원 아이디로 SSE로 연결을 한  후 댓글과 좋아요를 누르면 아래와 같은 결과가 나옵니다.