SSE 적용 (1) - SSE를 적용해보자 개념

배치 동작 완료 시점마다 화면에 update를 해줘서 사용자가 최신 결과를 바로 확인할 수 있도록 서비스를 제공하기위해 

적절한 기술 검토중에 SSE 추천을 받아 분석 후 개발을 진행하게 되었다.

관련 내용을 정리

 

SSE란, Server Sent Event

웹 브라우저와 서버 간 단방향 통신을 가능하게 하는 웹 기술

서버에서 클라이언트(브라우저)로 실시간 이벤트를 보낼 수 있다.

클라이언트 측에서 서버로의 연결을 열어두고 서버에서 이벤트를 푸쉬하는 방식

서버는 이벤트를 생성하고, 이벤트에 필요한 Data를 포함하여 Client에게 전달한다.

Client는 이벤트 수신 후 필요한 작업을 수행하는데 데이터를 사용

 

SSE 특징

- 단방향 통신 서버 -> 클라이언트로만 Data 전송

- 기본 HTTP 프로토콜 사용 -> 추가 Library나 프레임워크 없이도 쉽게 구현이 가능하다.

- 실시간으로 서버에서 데이터를 수신하여 동적으로 업데이트가 가능함

- client는 same sse emitter를 공유할 수 없다. 구독당 하나의 emitter를 만들어야한다. (동일 브라우저든 다른 user든 어디서든 독립적으로 유지가 됨)

- 브라우저 최대 동시접속수가 http1.1의 경우 6개, http2의 경우 100개 까지 가능하다.

- 클라이언트에서 emitter를 close해도 서버에서 감지하기 어렵다. (단방향 단점)

 

Sample 구현

- 진짜 Sample Level에서 조금 손봄

- API, Client, Batch(특정 작업 완료 후 broadcast 호출을 위함)

- 기본 핵심 구성 Subscribe, Broadcast 정도만

 

API

구성

- subscribe(api)

    - 구독 요청 -> emitter 생성 -> client에 생성이 완료되었다는 초기 데이터 응답 -> 완료

- broadcast(mq)

    - 단일 서버라면 API에 놓고 생성해도 상관이 없지만 이중화 된 서버이기 때문에 어디서 생성을 했던 이슈가 없도록 MQ를 적용

    - 특정 작업 완료 이후 broadcast 요청 -> 매핑 이벤트 찾아 응답 -> 완료

 

코드 샘플

1. 구독 API

//Controller

@GetMapping(value = {"/api/sse/subscribe/{orgId}/{menuName}"}, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter subscribeSse(
        @PathVariable String orgId,
        @PathVariable String menuName) {
    return sseService.subscribe(orgId, menuName);
}
 
 
//Service

private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
private static final long SSE_TIMEOUT_LIMIT_MINUTE = 3;
private final RabbitTemplate rabbitTemplate;
 
public SseEmitter subscribe(String orgId, String menuName) {
    SseEmitter emitter = new SseEmitter(TimeUnit.MINUTES.toMillis(SSE_TIMEOUT_LIMIT_MINUTE));
    String eventId = getEventIdPrefix(cmpnId, menuName) + "_" + Utils.getUuid();
    emitters.put(eventId, emitter);
 
    emitter.onCompletion(() -> emitters.remove(eventId));
    emitter.onTimeout(emitter::complete);
    emitter.onError(error -> {
        log.error("[Sse Error] eventId : {}, cause: {}", eventId, error.getMessage());
        emitter.complete();
    });
 
    sendEventToClient(emitter, eventId, menuName, SseExecutionStatus.EVENT_CREATED.getValue());
    return emitter;
}
 
private String getEventIdPrefix(String orgId, String menuName) {
    StringJoiner eventIdBuilder = new StringJoiner("_");
    eventIdBuilder.add(orgId)
                  .add(menuName);
    return eventIdBuilder.toString();
}
 
private void sendEventToClient(SseEmitter emitter, String eventId, String menuName, String data) {
    try {
        emitter.send(
                SseEmitter.event()
                        .id(eventId)
                        .name(menuName)
                        .data(data));
    } catch (Exception e) {
        emitter.completeWithError(e);
    }
}

 

2. 이벤트 Broadcasting API

// Service (mq 사용하여 따로 API 없이 consumer 부분)
// getEventIdPrefix, sendEventToClient는 위와 동일 

public void broadcastToMappedEvent(SseBroadcastModel message) {
    String idPrefix = getEventIdPrefix(message.getCmpnId(), message.getMenuName());
 
    emitters.entrySet().stream()
            .filter(entry -> entry.getKey().startsWith(idPrefix))
            .forEach(entry -> {
                SseEmitter emitter = entry.getValue();
                String eventId = entry.getKey();
                sendEventToClient(emitter, eventId, message.getMenuName(), eventId + "_" + message.getSubId()+ "_" + message.getStatus());
            });
}

 

Client

구성

- sse 생성 필요 메뉴 진입 시 event 생성

- vue store에서 이벤트 생성 및 저장

- api에서 broadcast를 통해 response data가 업데이트 되면, 각 화면에서 필요 작업 수행

 

코드 샘플

1. 화면

...
created: function() {
    ...
    this.initSse();
},
computed: {
    ...mapGetters({
        ....
        sseEventSource: 'sse/sseEventSource',
        sseResponseData: 'sse/sseResponseData',
        isSseEventSourceClosed: 'sse/isSseEventSourceClosed',
        lastEventId: 'sse/lastEventId'
    }),
},
watch: {
    ...
    sseResponseData(newVal) {
        if (newVal.indexOf('_complete') !== -1) {
            this.getDataList();
        }
    },
    isSseEventSourceClosed: function() {
       this.getDataList();
    }
}
methods: {
    ...,
    initSse() {
        let payload = {
            orgId : this.orgId,
            menuName : 'sampleMenu'
        }
        if (!this.sseEventSource || (this.orgId !== this.getOrgIdInLastEventId(this.lastEventId))) {
            this.$store.dispatch('sse/completeSseEventSource');
            this.$store.dispatch('sse/createSseEventSource', payload);
        }
   },
   getOrgIdInLastEventId(lastEventId) {
       return lastEventId.substring(0, lastEventId.indexOf('_'));
   }
}

 

2. Store

//actions.js
 
const createSseEventSource = (context, payload)  => {
    let url = `${UrlsConfig.govUrl}/api/sse/subscribe/${payload.orgId}/${payload.menuName}`;
    let eventSource = new EventSource(url);
 
    eventSource.addEventListener(payload.menuName,event => {
        context.commit('setLastEventId', event.lastEventId);
        context.commit('setSseResponseData', event.data+'_'+event.timeStamp);
    })
    eventSource.onerror = (event) => {
        if (event.target.readyState === EventSource.CLOSED) {
            context.commit('setIsSseEventSourceClosed', true);
            console.error('SSE CREATE ERROR : ' + '(' + event + ')');
        }
 
    }
    context.commit('setSseEventSource', eventSource);
}
 
const completeSseEventSource = (context) => {
    if (!context.state.sseEventSource) {
        return;
    }
 
    context.state.sseEventSource.close();
    context.commit('setSseEventSource', null);
 
    let url = `/api/sse/unsubscribe/${context.state.lastEventId}`;
    try {
        axios.delete(url);
    } catch (err) {
        console.error("SSE COMPLETE ERROR : ", err);
    } finally {
        context.commit('setLastEventId', '');
    }
}
 
 
//getters.js
 
const sseEventSource = (state) => state.sseEventSource;
const sseResponseData = (state) => state.sseResponseData;
const isSseEventSourceClosed = (state) => state.isSseEventSourceClosed;
const lastEventId = (state) => state.lastEventId;
 
 
//index.js
 
const state = {
    sseEventSource: null,
    lastEventId: '',
    sseResponseData: null,
    isSseEventSourceClosed: false
}
 
 
//mutations.js
 
const setSseEventSource = (state, sseEventSource) => {
    state.sseEventSource = sseEventSource;
};
 
const setSseResponseData = (state, sseResponseData) => {
    state.sseResponseData = sseResponseData;
};
 
const setIsSseEventSourceClosed = (state, isClosed) => {
    state.isSseEventSourceClosed = isClosed;
}
 
const setLastEventId = (state, lastEventId) => {
    state.lastEventId = lastEventId;
}

Batch

역할

- 특정 작업 완료 시 각 화면에서 필요한 작업을 수행하기 위해 Broadcast로 메시지 전달(Trigger)

- 실제 Broadcast 자체는 배치가 아닌 API에서 이루어짐 (배치는 트리거만 + 이중화 위해 MQ 사용)

public void sseBroadcast(String orgId, SseExecutionStatus status) {
    SseBroadcastModel sseBroadcastModel = SseBroadcastModel.builder()
            .cmpnId(cmpnId)
            .menuName(menuName)
            .status(status.getValue())
            .build();
 
    rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_SSE_BROADCAST, "", sseBroadcastModel);
}