Nuwa Project - SSE(Server-Side-Event) Connect

반응형

SMALL

현재 STOMP와 SockJS를 사용하여 채팅을 구현하고 있습니다.

채팅을 구현을 하던 도중 알림 기능도 추가가 되어야 하는 것 아닌가란 얘기가 나왔습니다.

이 부분에 대해서 고민을 하던 도중 SSE를 사용하여 구현을 하기로 했습니다.

 

웹소켓은 처음 연결을 진행하고 연결이 끊기기 전까지 지속적으로 양방향 통신이 가능합니다.

웹소켓으로 알림 구현을 생각하면 어플리케이션의 전반적인 웹소켓을 로그인 시점에 열어두고

해당 웹소켓을 이용하여 이벤트가 일어나면 해당 웹소켓을 이용하여 알림을 보내주는 방식을 구현 해볼까 생각을 했지만

알림을 위해 전반적으로 웹소켓을 열어두고 사용하는 것은 많은 비용이 소모가 되지 않을까 생각했습니다.

서버에서 특정 이벤트가 생기면 해당 이벤트를 전송하는 방식이 비용 소모가 적지 않을까란 생각을 했고

제 생각과 어울리는 기술은 SSE라고 생각하였고 구현을 진행 했습니다.

 

전반적인 구현은 아직 구현 중이고 먼저 connect를 하는 과정만 다룰 예정입니다.

 

SSE (Server-Side-Evnet)

SSE를 사용할 때 먼저 구독과 같이 연결을 진행을 한 상태에서 해당 연결이 지속이 될 때 서버에서 이벤트가 발생이 되면

응답 값을 내려주는 방식입니다.

 

이벤트가 오고 가는 타입은 text/event-stream 입니다.

 

EmitterRepository

@Repository
public class EmitterRepository {

    //Map에 회원과 연결된 SSE SseEmitter 객체를 저장
    public final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    //Event를 캐시에 저장
    private final Map<String, Object> eventCache = new ConcurrentHashMap<>();

    // id와 sseEmitter를 매개변수로 받아 emitters 맵에 저장
    public SseEmitter save(String id, SseEmitter sseEmitter) {
        emitters.put(id, sseEmitter);
        return sseEmitter;
    }

    //  emitters 맵에서 해당 id를 가진 항목을 삭제
    public void deleteById(String id) {
        emitters.remove(id);
    }
}

 

SSE SseEmitter 객체를 저장하는 Repository입니다.

추후에 Last-Event-Id 값을 받아와서 미수신된 데이터를 맵에서 찾아와서 추가로 전송을 해주는 로직을 작성을 할 예정입니다.

 

NotificationService.java

@Slf4j
@Service
@RequiredArgsConstructor
@Transactional(readOnly = true)
public class NotificationService {

    private final NotificationRepository notificationRepository;
    private final EmitterRepository emitterRepository;
    private final UserRepository userRepository;
    private final JwtUtil jwtUtil;

    // 29분
    private static final Long DEFAULT_TIME_OUT = 1000L * 60 * 29;

    // SSE 연결
    @Transactional
    public SseEmitter subscribe(String accessToken) {

        String findEmail = jwtUtil.getEmail(accessToken);

        // 멤버 찾기
        User findUser = userRepository.findByEmail(findEmail)
                .orElseThrow(() -> new NotFoundException(USER_NOT_FOUND));

        Long findUserId = findUser.getId();

        // Emitter Id
        String emitterId = findUserId + "_" + System.currentTimeMillis();

        // Emitter Id와 29분의 타임아웃을 가진 emitter를 생성 후 map에 저장
        SseEmitter saveEmitter = emitterRepository.save(emitterId, new SseEmitter(DEFAULT_TIME_OUT));

        log.info("new Emitter = {}", saveEmitter);

        // 상황 별 emitter 삭제
        // 완료
        saveEmitter.onCompletion(() -> emitterRepository.deleteById(emitterId));
        // 타임아웃
        saveEmitter.onTimeout(() -> emitterRepository.deleteById(emitterId));

        // 503 에러를 방지한 더미 데이터 전송
        dummyDateSend(saveEmitter, emitterId, "Event Stream Created. [memberId =" + findUserId +"]");

        return saveEmitter;
    }

    // 더미 데이터 반환 (503 Service Unavailable 방지)
    private void dummyDateSend(SseEmitter sseEmitter, String emitterId, Object data) {
        try {
            sseEmitter.send(SseEmitter.event()
                    .id(emitterId)
                    .name("sse")
                    .data(data)
                    .build());
        } catch (IOException e) {
            emitterRepository.deleteById(emitterId);
            log.error("SSE 연결 오류", e);
        }
    }

}

 

토큰 값을 가져와서 해당 User의 ID 값을 찾아줍니다.

그 후에 emitter id를 해당 userId와 현재 시스템 시간을 더해 생성을 합니다. (추후 미수신 데이터를 위해 현 시각을 붙여서 생성)

그리고 SSE는 처음 연결 시 데이터가 없으면 503 Service Unavailable 오류가 발생합니다.

오류가 발생하지 않게 더미 데이터를 넣어서 연결 시 보내줍니다.

 

NotificationController.java

@Slf4j
@RequestMapping("/notification")
@RestController
@RequiredArgsConstructor
public class NotificationController {

    private final NotificationService notificationService;
    private final CommonService commonService;

    @GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public ResponseEntity<SseEmitter> subscribe(@RequestHeader("Authorization") String accessToken) {
        SseEmitter emitter = notificationService.subscribe(accessToken);

        return ResponseEntity.status(OK).body(emitter);
    }
}

 

해당 컨트롤러에서 Service에서 작성한 emitter를 전송합니다.

이 부분에서 ResponseEntity에 ResponseDto를 생성하여 emitter를 넣어 보내는 방법을 하려고 했는데

타입이 text/event-stream이여서 convert 오류가 발생을 했습니다.

 

오류 내용

No converter for [class org.springframework.web.servlet.mvc.method.annotation.SseEmitter] with preset Content-Type 'null

 

현재와 같이 SseEmitter로 반환을 해주면 오류가 나지 않고 테스트가 가능합니다.

 

해당 값을 Application/json으로 반환은 할 수 없을 것 같고

추후에 Notification용 dto를 생성해서 해당 값을 json으로 넣어서 Ssemitter로 반환하면 되지 않을까 싶습니다.