SSE 적용 (3) - SSE 적용 시 이슈 처리

1. Event Share 이슈 처리 (동시 접속 고려)

이슈
- 여러개의 탭을 띄워놨을 때 Emitter Subscribe Response는 다 OK고 Event가 제대로 생성었지만 가장 최근에 생성된 탭만 event data 업테이트가 됨
원인 
- API에서 동일 이벤트가 아닌 다음 생성된 동일 키의 이벤트로 덮어 쓰기가 되어져버림
- = API에서 concurrentHashMap에 emitter 담고 있는데 동일 key로 새 것으로 엎어쳐져서
해결 방법
- emitter는 공유 안됨
- 이벤트 구독이 필요할 때마다 여러개 emitter를 uuid 추가 및 idPrefix적용해서 Broadcast 시 idPrefix 하위 것들 모두 업데이트 되도록 변경
 

2. 인증

이슈
- event subscribe시 auth를 추가할 수 없음
원인 
- 인증을 위해 헤더에 토큰을 전달해야하는데 기본 EventSource에는 header 추가가 안되어버림
해결 방법
a. EventSourcePolyfill 라이브러리 사용해서 헤더에 Authorization 추가
  - 예제 코드 

let tokenKey = Profile.env==='PRODUCTION' ? VueCookies.get(GlobalConstants.ACCESS_TOKEN) : VueCookies.get(GlobalConstants.ACCESS_TOKEN_DEV);
const EventSource = EventSourcePolyfill || NativeEventSource;
this.eventSource = new EventSource(
    `${UrlsConfig.menu1Url}/api/sse/subscribe/${this.currentCmpnId}/gridList`,
    {
        headers: {
                    Authorization: `Bearer ${tokenKey}`,
                  },
        heartbeatTimeout: 12 * 60 * 1000,
    }
);

  - 이슈
    i) 검사 완료시 브로드캐스트는 정상 동작(401 안남) 이슈가 개발자 도구에서 EventStream 업데이트 되는 데이터가 안보임(pending은 아닌데 안뜸 )
    ii) https://github.com/Yaffle/EventSource/issues/79
  - -> 롤백
b. 구독 시 인증 제외
- 어떤 고객이 다른 회사의 다른 고객의 아이디 모두를 알아서 이벤트를 훔쳐본다고 했을때 유출이 될 만한 정보
  - 이벤트 스트림 무단 구독을 통해서 뭔지 모를 검사가 끝났다는 사실
  - companyID, dataID
- 크게 의미있는 데이터 X → 결론 : Subscribe시에는 인증 제외하도록 구현
 

3. 프록시 통한 API Call로의 변경

API GW 설정이 안되어있기 때문에 프로젝트 프록시 통해서 ClientUrl → BackendUrl로 변경해서 API 콜 해야함
이슈
- 프록시 통해서 clientURL로 subscribe 진행시 header connection reponse close error → pending
해결 방법
- webpack → compress: false 옵션 변경
 

4. http 1.1 브라우저 제한 이슈

이슈
- http1.1 사용 시 브라우저당 6개 eventSource를 생성 가능
  - 이미 emitter에 들어있는 이벤트 객체가 6개가 초과한다고 하더라도 브라우저가 6개 미만이면 상관없음
해결 방법
- 웹 서버(예: Apache, Nginx)에서는 HTTP/2.0를 지원, 이를 이용하여 특정 경로 또는 URL에 대해서만 HTTP/2.0을 활성화하는 방법
- id에 miliseconds를 추가하여 maximum이 되는 경우 최초 것부터 삭제하도록 변경
  - user가 추가되어야해서 URL이 길어짐
- → 복잡하기도 하고 다른 유저가 아닌 한 개인이 여러 브라우저를 띄우는거라 큰 이슈로 보여지진않아서 우선은 Timeout을 적절하게 줘서 빠른 해지 빠른 reconnect(새 탭에서 먼저 가져갈 수 있도록)
 

5. SEEmitter 공통화

이슈
- Event list, detail 나눠놓음 → 분기가 많이 생김
- detail 검사 진행 시 검사시마다 이벤트가 새로 생성됨
해결 방법
- vue store 사용하여 하나의 유저가 Menu1 메뉴에서 emitter는 하나로 계속 사용하도록 처리
 

6. net::ERR_HTTP2_PROTOCOL_ERROR 200 (OK) 및 여러 Nginx 이슈

이슈
- subscribe 시 pending 이슈 
- nginx 사용 개발 적용 시 1분 주기로 client에서 sse 접속 재시도 및 Error 발생 이슈
해결 방법
- 별도 리트라이 없어도 client쪽에서는 연결이 끊어지면 sse 자동으로 호출하는 것으로 보임 (=api로직의 retry 제거 및 client proxy_read_timeout 360; 추가 제공)
- api 에서 타임아웃은 client보다 먼저 발생시켜서 client가 자연스럽게 재시도하도록 설정 (api timeout 3m)
- SSE관련 Nginx 설정들

location /api/sse/ {
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    proxy_pass http://${api}:{port}$request_uri;
    proxy_read_timeout 360;
    charset utf-8;
}

 

7. 메모리 이슈 방지 (client, server lifecycle 고려)

이슈
- 이벤트 발생 페이지를 벗어나 다른 메뉴로 가는 경우 이벤트가 유지되어 장기적으로 보면 메모리 이슈 발생 가능성이 있음
- client life cycle과 server lifecycle을 맞추지 않으면 어디선가 이벤트 해지되지 않아 누수 발생 가능성
해결 방법
- 기본적으로 client와 server 두 사이클을 맞추긴 어려움 → 적절한 timeout 주기 + 메뉴 벗어나는 경우 client, server의 이벤트 명시적 해지
- API onComplete 부분 수정
  a. 이중화 서버별 remove로 작업 진행 → remove가 아닌 complete로 변경
  b. 서버쪽에서 sse 이벤트 처리가 완료되면 sse emitters에서 해당 emitter 삭제하는게 아니고 complete 쳐서 프로세스를 태워야 정상적으로 종료
  c. 직접 삭제는 onComplete 내부에서만 하면 됨
- sse client 이벤트 초기화 및 unsubscribe 구현 (mq)
  - Flow
    - 이벤트 생성 외 페이지에서는 페이지 진입 시 store에 담긴 event 초기화 / 이벤트 생성 페이지는 event 초기화 및 신규 생성
    - client에서 event close및 store 초기화 진행하면서 api에 해지 요청을 날림 
    - unsubscribe api에서 complete 수행
  - Client
    - 예제 코드 

//list.vue

...
initSse() {
    let payload = {
        cmpnId : this.currentCmpnId,
        menuName : 'menu1'
    }
    if (!this.sseEventSource || (this.currentCmpnId !== this.getCmpnIdInLastEventId(this.lastEventId))) {
        this.$store.dispatch('sse/completeSseEventSource'); //해지
        this.$store.dispatch('sse/createSseEventSource', payload); //생성
    }
},
 
...
 
 
//actions.vue
import axios from 'axios';
...
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', '');
    }
}
...

  - API
    - 예제 코드 

//Controller
@DeleteMapping(value = "/api/sse/unsubscribe/{eventId}")
    public ResultModel removeDestroyedEmitter (@PathVariable("eventId") String eventId) {
        return sseService.sseUnsubscribe(eventId);
}
 
 
//Service
//이중화로 어느 서버에 해당 이벤트가 있을지 모르니 MQ로 처리
public ResultModel sseUnsubscribe(String eventId) {
    ResultModel result = new ResultModel();
    try {
            rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_SSE_UNSUBSCRIBE_BROADCAST, "", eventId);
        } catch (Exception e) {
            result.setError("500", e.getMessage());
            result.setStatus("fail");
        }
        return result;
 
}
 
 
//RabbitMqConfig
public static final String EXCHANGE_SSE_UNSUBSCRIBE_BROADCAST = "fanout.sse.unsubscribe.broadcast";
public static final String QUEUE_SSE_UNSUBSCRIBE_BROADCAST_PROCESS = "sse.unsubscribe.broadcast.process";
 
@Bean
public FanoutExchange sseBroadcastUnsubscribeFanoutExchange() {
    return new FanoutExchange(EXCHANGE_SSE_UNSUBSCRIBE_BROADCAST);
}
 
 
@Bean
public Queue getSseUnsubscribeQueue() {
    return new Queue(QUEUE_SSE_UNSUBSCRIBE_BROADCAST_PROCESS);
}
 
@Bean
public Binding bindingSseUnsubscribeExchangeAndQueue(Queue getSseUnsubscribeQueue, FanoutExchange sseBroadcastUnsubscribeFanoutExchange) {
    return BindingBuilder.bind(getSseUnsubscribeQueue).to(sseBroadcastUnsubscribeFanoutExchange);
}
 
 
//Consumer
@RabbitListener(queues = RabbitMqConfig.QUEUE_SSE_UNSUBSCRIBE_BROADCAST_PROCESS, concurrency = "${consumer.sse-unsubscribe-broadcast}")
public void sseUnsubscribeReceiveMessage(String eventId) {
    sseService.broadcastToUnsubscribe(eventId);
}
 
 
//Service
public void broadcastToUnsubscribe(String eventId) {
    try {
        SseEmitter emitter = emitters.get(eventId); //subscribe시 map에 넣어놨던 emitter
        if (!ObjectUtils.isEmpty(emitter)) {
            emitter.complete();
        }
 
    } catch (Exception e) {
        log.error("[Sse unsubscribe Error] cause : {}, eventId : {}", e.getMessage(), eventId);
    }
}

- 다른 서비스로 갈때에는 타임아웃 이후 자동으로 이벤트 해지 됨 (재연결안함)