SseEmitter

Spring Framework에서는 SSE 기능을 편리하게 사용할 수 있도록 SseEmitter라는 구현체를 지원한다.

회원 중복 연결 제어

기존에는 EmitterRepository를 아래와 같이 단순히 member.id 기준으로 다뤘다.

private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();

이렇게 할 경우 다음과 같은 문제가 발생한다.

회원 1 - 탭 A 연결
회원 1 - 탭 B 연결 → 탭 A가 덮어써짐
탭 A 종료 콜백 실행 → 회원 1의 탭 B emitter까지 삭제

그렇기에 SseEmitter와 각각의 고유 ID를 매핑하여 아래와 같이 구조로 emitter들을 관리한다.

private final ConcurrentMap<Long, ConcurrentMap<String, SseEmitter>> emitters = new ConcurrentHashMap<>();
 
public record EmitterConnection(  
        String emitterId,  
        SseEmitter emitter  
) {  
}

SUBSCRIBE 이벤트는 모든 client에게 보낼 필요 없기에 memberId + emitterId 조합의 알림 전송이 필요함.

에러 제어

emitter를 생성 후 콜백 함수를 등록한다. 보통 emitter 객체를 삭제하는 방식이다.

// 사용자 ID를 기반으로 Emitter 생성  
private EmitterConnection createEmitter(Long memberId) {  
    SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);  
    String emitterId = createEmitterId();  
    emitterRepository.save(memberId, emitterId, emitter);  
  
    // 콜백 등록  
    Runnable cleanup = () -> emitterRepository.delete(memberId, emitterId, emitter);  
  
    // SSE 연결이 정상적으로 종료되었을 때 실행  
    emitter.onCompletion(cleanup);  
    // DEFAULT_TIMEOUT을 초과해서 연결이 타임아웃 되었을 때  
    emitter.onTimeout(cleanup);  
    // SSE 연결 중 에러가 발생했을 때  
    emitter.onError(error -> cleanup.run());  
  
    return new EmitterConnection(emitterId, emitter);  
}

IllegalArgumentException를 에러 포함시킨 이유는 다음과 같다.

등록된 HttpMessageConverter들을 돌면서 data를 쓸 수 있는 converter를 찾습니다. 아무 converter도 처리할 수 없다면 Spring이 직접 다음과 같이 IllegalArgumentException을 던진다.

private void sendLogic(EmitterConnection connection, Long memberId, SseEventType eventType, Object data) {
	String emitterId = connection.emitterId();
	SseEmitter emitter = connection.emitter();
	try {
		emitter.send(
				SseEmitter.event()
						.id(String.valueOf(memberId))
						.name(eventType.name())
						.data(data)
		);
	} catch (IOException | IllegalArgumentException e) {
		emitterRepository.delete(memberId, emitterId, emitter);
		emitter.completeWithError(e);
	}
}

이벤트 기반으로 트랜잭션 분리

트랜잭션 내에 SSE 전송을 포함시키면 다음과 같은 문제점이 발생한다.

DB 트랜잭션의 생명주기에 네트워크 I/O가 포함된다.
예외가 강하게 결합된다.
트랜잭션은 취소되지만 SSE 메세지는 전송될 수 있다.

또한 OSIV가 켜져있을 시 SSE 서비스단에 트랜잭션이 걸려있다면 SSE연결 동안 트랜잭션을 계속 물고 있어 커넥션 낭비가 일어날 수 있으니 트랜잭션을 걸지 않는 것을 권장한다.

이벤트 객체 정의하고 EventListener Bean 등록한다.
이벤트 리스너 정의 및 @TransactionalEventListener 어노테이션 추가한다.
발생 시점은 COMMIT 이후로 한다.

@Component  
@RequiredArgsConstructor  
public class AlarmEventListener {  
  
    private final AlarmChannelService alarmChannelService;  
  
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)  
    public void notifyAlarm(AlarmCreateEvent event) {  
        alarmChannelService.notify(event.receiverId(), event.alarm());  
    }  
}

그리고 ApplicationEventPublisher를 통해 이벤트를 발행한다.

private final ApplicationEventPublisher applicationEventPublisher;
 
@Transactional  
public Alarm create(Long receiverId, AlarmType type, String title, String content, String targetUrl) {  
    Member receiver = memberService.getMember(receiverId);  
    Alarm alarm = Alarm.create(receiver, type, title, content, normalizeTargetUrl(targetUrl));  
    applicationEventPublisher.publishEvent(new AlarmCreateEvent(receiverId, alarm));  
    return alarmRepository.save(alarm);  
}

하트비트

SSE는 일반 HTTP 응답과 달리 응답을 끝내지 않고 연결을 유지한다.
이때 클라이언트가 연결을 끊어도 Servlet 서버가 바로 알 수 없다.

그렇기에 아래 Spring 공식 문서에서도 연결이 끊어질때 쓰기 작업이 실패하므로 주기적으로 데이터를 전송하는 방식을 권장한다. 그리고 해당 방식을 heartbeat라고 부른다.

Quote

The Servlet API does not provide any notification when a remote client goes away. Therefore, while streaming to the response, whether through SseEmitter or reactive types, it is important to send data periodically, since the write fails if the client has disconnected. The send could take the form of an empty (comment-only) SSE event or any other data that the other side would have to interpret as a heartbeat and ignore.

Alternatively, consider using web messaging solutions (such as STOMP over WebSocket or WebSocket with SockJS) that have a built-in heartbeat mechanism.

SSE 연결은 장시간 데이터가 흐르지 않으면 프록시, 로드밸런서 등의 idle timeout에 의해 연결이 종료될 수 있다.
따라서 실무에서는 연결을 유지하고 끊어진 연결을 감지하기 위해 주기적으로 heartbeat를 전송할 수 있다.

SseEmitter에서는 comment()를 이용해 SSE의 comment 형식으로 heartbeat를 전송할 수 있다.
또한 heartbeat 요청을 지속적으로 보내야 하기에 아래와 같이 스케줄러를 활용할 수 있다.

@Scheduled(fixedRateString = "PT15S")  
public void sendHeartbeat() {  
    emitterRepository.findAll().forEach(connection ->  
            send(connection, SseEmitter.event().comment("heartbeat"))  
    );  
}