TIL SSE 알림 기능 만들기 총정리 230201

SSE(Server-Sent-Event)를 사용해서 알림기능을 만들기로 결정을 하고 약 2주동안 고생하면서 결국에는 성공(?)을 시켰습니다. 완전하다고는 할 수 없지만 이제는 왠만하면 SSE를 사용하지는 않고 다른 옵션을 시도해볼 것 같습니다. 로컬에서는 잘 돌아갔는데 프론트랑 연결하면서 팡팡팡팡팡! 에러가 하루가 멀다하고 터졌거든요. 산넘어 산이었습니다. 이번 포스팅에서는 다른 블로그들은 언급하지 않고 넘어가는 SSE 알림 기능 관련 에러 총정리를 하려고 합니다. 저처럼 헤매지 마시길.

 

SSE 알림 기능 관련 주요 TIL

230110 알림 기능 구현 뭘로 할까?

230111 SSE로 알림기능 구현하기

230120 SSE 프론트와 연결하기 (feat. 비관적락)

230124 SSE 에러: DB connection leak, open-in-view 설정

230125 SSE 또 connection leak triggered

230126 @Async 비동기처리

230126 읽지 않은 알림 갯수 반환하기

230127 @EventListener 알림 기능 강결합 제거

230131 AsyncConfigurerSupport로 비동기 설정하기

 

 

 

위 처럼 다양한 에러를 만나고 결국엔 해결했습니다. TIL에서 언급하지 않은 부분과 그래서 결국 최종 코드가 무엇인지에 대해 알려드리도록 하겠습니다. 하나 걸러 하나가 나오는 에러라 다 기록하지는 못했지만 앞으로는 더 꼼꼼히 기록해야겠다는 다짐을 하면서 우선 SSE 설정을 위한 헤더에 대해 알아봅시다.

 

NotificationController

@Tag(name = "SSE")
@ApiResponses(value = {
        @ApiResponse(responseCode = "2000", description = "SSE 연결 성공"),
        @ApiResponse(responseCode = "5000", description = "SSE 연결 실패")
})
@Operation(summary = "SSE 연결")
@GetMapping(value="/api/subscribe/{nickname}", produces = "text/event-stream")
public SseEmitter subscribe(
        @PathVariable String nickname,
        @RequestHeader(value="Last-Event-ID", required = false, defaultValue = "") String lastEventId,
        HttpServletResponse response){
    response.addHeader("X-Accel-Buffering", "no");
    response.addHeader("Content-Type", "text/event-stream");
    response.setHeader("Connection", "keep-alive");
    response.setHeader("Cache-Control", "no-cache");

    String encodedNickname = URLDecoder.decode(nickname, StandardCharsets.UTF_8);
    return notificationService.subscribe(lastEventId, encodedNickname);
}

Ngnix를 적용하고 에러가 나서,

 

response.addHeader("X-Accel-Buffering", "no");

를 적용해 주었고

 

아래와 같이 에러가 나서

EventSource's response has a Content-type specifying an unsupported type: application/json. Aborting the connection

EventSource's response has a Content-type specifying an unsupported type: application/json. Aborting the connection

response.addHeader("Content-Type", "text/event-stream");

 

를 적용해주었습니다.

 

ngnix 세팅 때문에 토큰을 헤더에 담아오지 못하게 되어서 nickname으로 파라미터를 변경해주었는데, 닉네임에 한글도 있어서 encoding관련해서 아래처럼 설정을 넣어주었습니다.

 

String encodedNickname = URLDecoder.decode(nickname, StandardCharsets.UTF_8);

 

 

NotificationService

    public SseEmitter subscribe(String lastEventId, String nickname) {
        Member member = memberRepository.findByNickname(nickname)
                .orElseThrow(()-> new NotFoundException(SSE, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + nickname));
        Long memberId = member.getId();
        String emitterId = memberId + "_" + System.currentTimeMillis();
        SseEmitter emitter = emitterRepository.save(emitterId, new SseEmitter(DEFAULT_TIMEOUT));

        log.info("emitter created");

        emitter.onCompletion(() -> {
            synchronized (emitter){
            emitterRepository.deleteById(emitterId);}});
        emitter.onTimeout(() -> {
            emitter.complete();
            emitterRepository.deleteById(emitterId);});

        sendToClient(emitter, emitterId, "EventStream Created. [memberId=" + memberId + "]");

        if (!lastEventId.isEmpty()) {
            Map<String, Object> events = emitterRepository.findAllEventCacheStartWithByMemberId(String.valueOf(memberId));
            events.entrySet().stream()
                    .filter(entry -> lastEventId.compareTo(entry.getKey()) < 0)
                    .forEach(entry -> sendToClient(emitter, entry.getKey(), entry.getValue()));
        }

        return emitter;
    }

    @Async // 비동기 처리를 위한 어노테이션
    @Transactional
    public void send(Member receiver, Member sender, NotificationType notificationType, String content, RedirectionType type, Long typeId, Long postId) {
        Notification notification = notificationRepository.save(new Notification(receiver, notificationType, content, type, typeId, postId, sender));
        String memberId = String.valueOf(receiver.getId());

        Map<String, SseEmitter> sseEmitters = emitterRepository.findAllEmitterStartWithByMemberId(memberId);
        sseEmitters.forEach(
                (key, emitter) -> {
                    emitterRepository.saveEventCache(key, notification);
                    sendToClient(emitter, key, notification.getContent());
                }
        );
    }

    private void sendToClient(SseEmitter emitter, String emitterId, Object data) {
        try {
            log.warn("emitterId : " + emitterId);
            log.warn("data : " + data.toString());
            emitter.send(SseEmitter.event()
                    .id(emitterId)
                    .data(data));
            log.info(emitterId+"-emitter has been sent and completed");
        } catch (IOException exception) {
            log.error("Unable to emit");
            emitter.completeWithError(exception);
            emitterRepository.deleteById(emitterId);
        }
    }

    @Transactional
    public List<ResponseNotificationDto> getNotificationList(Member member) {
        List<Notification> notificationList = notificationRepository.findAllByReceiverOrderByCreatedAtDesc(member);
        List<ResponseNotificationDto> responseNotificationDtoList= new ArrayList<>();
        for (Notification notification: notificationList) {
            responseNotificationDtoList.add(SSE_MAPPER.NotificationtoResponseNotificationDto(notification));
        }
        return responseNotificationDtoList;
    }

    @Transactional
    public void readNotification(Long notificationid, Member member) {
        Notification notification = notificationRepository.findById(notificationid)
                .orElseThrow(()-> new NotFoundException(SSE, SERVICE, NOTIFICATION_NOT_FOUND, "Notification ID : " + notificationid));
        if(member.getId().equals(notification.getReceiver().getId())) {
            notification.read();
        }
        notificationRepository.save(notification);
    }

    @Transactional
    public ResponseCountNotificationDto countUnreadNotifications(Member member) {
        String nickname = member.getNickname();
        Long count = notificationRepository.countUnreadNotifications(nickname);

        return new ResponseCountNotificationDto(count);

    }
}

여러가지 메소드가 들어있어서 좀 깁니다. 

 

중요한 수정사항들을 몇가지 언급하자면,

 

synchornized() : 비동기를 사용했던 친구들이 완료되면 동기화 시키는 메서드

emitter.completeWithError(exception);

에미터가 발신 되지 않았을 경우 에러와 함께 종료 시키기

 

 

 

대댓글 알림 서비스 메서드 예시

제가 EventListener를 사용했다는 사실 알고 계시죠? (@EventListener 사용)  

아래와 같이 EventListener를 구현하였습니다.

@Component
@RequiredArgsConstructor
@Slf4j
public class NotificationListener {

    private final NotificationService notificationService;

    @TransactionalEventListener
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    @Async
    public void handleNotification(RequestNotificationDto requestNotificationDto){
        notificationService.send(requestNotificationDto.getReceiver(), requestNotificationDto.getSender(),requestNotificationDto.getNotificationType(),
                requestNotificationDto.getContent(), requestNotificationDto.getType(), requestNotificationDto.getTypeId(), requestNotificationDto.getPostId());
        log.info("EventListener has been operated. Sender Id: " + requestNotificationDto.getSender().getId() + " NotificationType: " +requestNotificationDto.getNotificationType());
    }

}

 

예시로는 대댓글을 작성하면 가는 알림입니다.

@Transactional
public void createComment(Long postId,Long parentId, CommentDto commentDto, String nickname) {

    Post post = postRepository.findById(postId).orElseThrow(
            () -> new NotFoundException(Domain.COMMENT, Layer.SERVICE, POST_NOT_FOUND, "Post ID : " + postId)
    );
    Comment parentComment = null;
    if (parentId != null) {
        parentComment = commentRepository.findById(parentId).orElseThrow(
                () -> new NotFoundException(Domain.COMMENT, Layer.SERVICE, COMMENT_NOT_FOUND, "Parent Comment ID : " + parentId)
        );
    }
    Member member = memberRepository.findByNickname(nickname)
            .orElseThrow(() -> new NotFoundException(COMMENT, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + nickname));

    Comment comment = COMMENT_MAPPER.commentDtoToComment(commentDto, member, post, parentId);
    commentRepository.save(comment);

    //post 작성자에게 댓글 알림
    Member postMember = memberRepository.findByNickname(post.getNickname())
            .orElseThrow(() -> new NotFoundException(COMMENT, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + nickname));

    if (!postMember.getNickname().equals(member.getNickname())) {
        String content = post.getTitle() + "에 " + nickname + "님이 댓글을 남겼습니다.";
        notify(postMember, member, NotificationType.COMMENT, content, RedirectionType.detail, postId, null);
    }

    //댓글 작성자에게 댓글 알림
    if (parentComment != null) {
        Member commentMember = memberRepository.findByNickname(parentComment.getNickname())
                .orElseThrow(() -> new NotFoundException(COMMENT, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + comment.getNickname()));
        if (!commentMember.getNickname().equals(member.getNickname())) {
            String content = commentMember.getNickname() + "님의 댓글에 " + nickname + "님이 댓글을 남겼습니다.";
            notify(commentMember, member, NotificationType.COMMENT, content, RedirectionType.detail, postId, null);
        }
    }
}


private void notify(Member postMember, Member sender, NotificationType notificationType,
                    String content, RedirectionType type, Long typeId, Long postId){
    eventPublisher.publishEvent(new RequestNotificationDto(postMember,sender, notificationType,content,type, typeId, postId));

}

위처럼 notify 라는 메서드를 만들어서 event를 발행합니다. 해당 event가 발행되면 listener가 듣고 NotificationService의 send 메서드를 실행시켜줍니다. 이때, Eventlistener의 파라미터는 1개이어야 하기 때문에 RequestNotificationDto를 생성해서 하나의 dto에 정보를 담아서 보내주었습니다.

 

 

 

SSE 에러를 해결하면서 느낀 점

인터넷에 나오는 모든 정보가 정확한 것은 아닙니다. 에러가 있어도 언급을 하지 않는 경우가 있고, 내 컴퓨터에서는 안되는 경우가 더러 있습니다. 그리고 동료들이 있어서 함께 에러를 해결해나갈수 있었습니다. github의 다른 개발자들도 같은 고민을 한 적이 있고 각기 다른 방법으로 에러를 해결했다는 사실도 재미있습니다. 

 

에러는 로컬에서 발생하는 에러(error factor : ONLY myself), 프론트와 연결하면서 발생하는 에러(error factor: still ONLY ME)가 있습니다. ngnix 설정을 하면서 또 한번 다수의 에러가 생겼고, 강결합으로 인한 에러도 있었습니다. 이번을 계기로 http 통신에 대해서 좀 더 알게되었습니다. 

 

 

아직도 미해결 에러가 있습니다. 해협!!!

많은 시간 공들여서 찾아보았지만 아직 해결되지 않은 에러가 한가지 남아있습니다.

 

바로바로.....

 

net:: ERR_INCOMPLETE_CHUNKED_ENCODING 200

 

입니다.

 

구글링으로 나와있는 방법은 대부분 시도해보았습니다. 혹시 해결방안을 알고 계시는 분들은 댓글달아주세요...!!

 

라고 하자마자 해결방안이 나와서 에러가 이제 안납니다.

 

ngnix 설정을 추가해 줍니다.

proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding off;

 

에러가 사라졌습니다.

위에 헤더로 ngnix 설정해줬다고 생각했었는데 아니었나봅니다....