SSE (Server Sent Events)

내가 담당했던 대시보드 기능 내에 실시간 알림 기능이 포함되어 있었다.

 

 

기존 HTTP 통신 방식은 1회성으로 요청(Request) - 응답(Response) 후 연결의 닫아버리는 특징을 가지고 있었기 때문에, 실시간 알림 기능 구현을 위해서는 클라이언트 - 서버 간 연결을 계속 유지 시킬 방법을 찾아야 했다.

 

방법으로 Web-Socket과 Server-Sent-Events 두 개의 선택지가 존재했다.

모두 실시간 통신을 가능하게 하지만 장단점이 존재했다.

 

출처 : https://surviveasdev.tistory.com/entry/%EC%9B%B9%EC%86%8C%EC%BC%93-%EA%B3%BC-SSEServer-Sent-Event-%EC%B0%A8%EC%9D%B4%EC%A0%90-%EC%95%8C%EC%95%84%EB%B3%B4%EA%B3%A0-%EC%82%AC%EC%9A%A9%ED%95%B4%EB%B3%B4%EA%B8%B0

 

실시간 알림 기능은 양방향 통신이 따로 필요없다는 점과, React - Spring에서 SSE에 관련된 함수가 내장되어 있다는 점, 별도의 세팅없이 HTTP 프로토콜을 사용해 통신한다는 점 때문에 SSE를 선택하여 실시간 알림 기능을 구현하기로 하였다.

 

전체 코드가 궁금하다면 GitHub를 참고하길 바란다.

GitHub - Bipum-In/Bipum-In-BE: 비품인 백엔드 레포입니다.

 

먼저, 클라이언트에서 로그인 시에 구독요청을 보내면 서버에서 SSEEmitter를 생성한다.

 

    private final NotificationService notificationService;

    @Operation(summary = "SSE 연결", description = "로그인 직후 SSE 연결해야합니다!")
    @GetMapping(value = "/subscribe", produces = "text/event-stream")
    public SseEmitter subscribe(@RequestHeader(value = "Last-Event-ID", required = false, defaultValue = "") String lastEventId,
                                @Parameter(hidden = true) @AuthenticationPrincipal UserDetailsImpl userDetails) {
        return notificationService.subscribe(userDetails.getUser().getId(), lastEventId);
    }

SSE 통신을 위해서 MIME 타입은 text/event-stream 으로 지정한다.

MIME 타입은 일종의 인코딩 방식으로 브라우저에서의

확장자 역할을 한다고 보면 된다. MIME 타입에 따라 브라우저는 URL을 처리하는 방법을 결정하게 된다.

 

Last-Event-ID 헤더는 클라이언트가 마지막으로 수신한 데이터의 id값을 의미하며, 모종의 이유로 연결이 끊어졌을 때 전달되지 못한 알림이 있을 때 해당 ID를 이용해서 전달되지 못한 알림들을 다시 보내줄 수 있다.

 

 

    //시간이 포함된 아이디 생성. SseEmitter 구분을 위함
    @Transactional
    public SseEmitter subscribe(Long userId, String lastEventId) {

        String emitterId = makeTimeIncludeId(userId);
        // lastEventId가 있을 경우, userId와 비교해서 유실된 데이터일 경우 재전송할 수 있다.

        emitterRepository.deleteAllEmitterStartWithId(String.valueOf(userId));

        SseEmitter emitter = emitterRepository.save(emitterId, new SseEmitter(DEFAULT_TIMEOUT));

        emitter.onCompletion(() -> {
            log.info("SSE 연결 Complete");
            emitterRepository.deleteById(emitterId);
        });
        //시간이 만료된 경우 자동으로 레포지토리에서 삭제하고 클라이언트에서 재요청을 보낸다.
        emitter.onTimeout(() -> {
            log.info("SSE 연결 Timeout");
            emitterRepository.deleteById(emitterId);
        });
        emitter.onError((e) -> emitterRepository.deleteById(emitterId));
        //Dummy 데이터를 보내 503에러 방지. (SseEmitter 유효시간 동안 어느 데이터도 전송되지 않으면 503에러 발생)
        String eventId = makeTimeIncludeId(userId);
        sendNotification(emitter, eventId, emitterId, "EventStream Created. [userId=" + userId + "]");

        // 클라이언트가 미수신한 Event 목록이 존재할 경우 전송하여 Event 유실을 예방한다.
        if (hasLostData(lastEventId)) {
            sendLostData(lastEventId, userId, emitterId, emitter);
        }

        return emitter;
    }

    private void sendNotification(SseEmitter emitter, String eventId, String emitterId, Object data) {
        try {
            log.info("eventId : " + eventId);
            log.info("data" + data);
            emitter.send(SseEmitter.event()
                    .id(eventId)
                    .data(String.valueOf(data)));

        } catch (IOException exception) {
            log.info("예외 발생해서 emitter 삭제됨");
            emitterRepository.deleteById(emitterId);
        }
    }

구독 요청을 받으면, 서버에서는 SSEEmitter를 생성해 저장하였다.

 

    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    private final Map<String, Object> eventCache = new ConcurrentHashMap<>();

    //save - Emitter를 저장한다. Emitter = 이벤트를 생성하는 메소드
    @Override
    public SseEmitter save(String emitterId, SseEmitter sseEmitter) {
        emitters.put(emitterId, sseEmitter);

        for(Map.Entry<String, SseEmitter> entrySet : emitters.entrySet()){
            log.info(entrySet.getKey() + " : " + entrySet.getValue());
        }
        return sseEmitter;
    }

Emitter는 클라이언트와 서버간 연결이 되어있을 때만 유지하면 되는 데이터이기 때문에 DB에 저장하지 않고, Map에 저장하여 관리하였다.

 

구독이 완료되면 지정된 시간동안은 연결이 계속 유지되며, SSE 기본 설정으로 연결이 만료되었을 시에 자동으로 클라이언트가 재요청을 보내 다시 연결을 유지한다.

알림을 보내고 싶은 로직에서 send메서드를 호출해서 원하는 데이터를 클라이언트에게 전달할 수 있다.

    @Secured(value = UserRoleEnum.Authority.ADMIN)
    @PostMapping(value = "/supply/excel", consumes = {MediaType.MULTIPART_FORM_DATA_VALUE})
    public ResponseDto<String> createSupplies(
            @ModelAttribute ExcelCoverDto excelCoverDto,
    @Parameter(hidden = true) @AuthenticationPrincipal UserDetailsImpl userDetails) throws IOException {

        List<Requests> requests = supplyService.createSupplies(excelCoverDto, userDetails.getUser());


        for(Requests request : requests){
            if (request.getSupply().getUser() != null) {
            	// 알림을 보내는 줄
                notificationService.sendForUser(userDetails.getUser(), request.getRequestId(), AcceptResult.ASSIGN);
            }
        }

        return ResponseDto.success("비품 등록 성공");
    }

    private String convertToJson(User sender, Notification notification) {
        String jsonResult = "";

        NotificationResponseDto notificationResponseDto = NotificationResponseDto.of(notification, sender.getImage());

        try {
            jsonResult = objectMapper.writeValueAsString(notificationResponseDto);
        } catch (JsonProcessingException e) {
            throw new CustomException(ErrorCode.JsonConvertError);
        }

        return jsonResult;
    }

단, SSE에서 데이터는 String 형식으로 한정된다. 따라서 Binary 타입일 경우 Json 으로 변환해서 클라이언트에게 전달해주었다.

 

알림을 보내도록 설정된 함수가 호출될 때, 알림도 함께 발송되게 된다.

 

SSE가 동작하는 과정을 간단히 요약하자면 다음과 같다.

 

1. 클라이언트가 서버로 구독요청을 보낸다.

2. SSEEmitter가 반환되고, 클라이언트 - 서버와의 연결이 유지된다.

3. 알림을 보낼 이벤트가 발생하면, 서버에서는 생성해둔 emitterId로 알림을 발송한다.

4. 로그아웃 시에 클라이언트에서 연결을 끊는다.

 

가장 요점적인 로직만 발췌하여 글에 담아두었으니, 만약 혼동된다면 글 상단에 첨부해둔 Github 링크를 참고하길 바란다.

만약, SSE를 사용할 예정인데 NGINX 서버를 함께 운용중이라면 추가적인 설정이 필요하니, 아래 글도 참고하면 좋을 것 같다.

 

항해 99 - 2023.03.28 TIL

 

항해 99 - 2023.04.24 TIL

 

 

 

참고 자료

[Spring + SSE] Server-Sent Events를 이용한 실시간 알림

[HTTP] 미디어 타입 (MIME Type)