구현

SSE
- 서버의 일방적인 푸시만으로도 알람 기능이 작동하므로 SSE 사용
- sse-starlette 라이브러리 → 비동기 프로그래밍 패러다임
router
- 연결된 유저 수만큼 무한 loop가 유지되며 Redis 메세지 큐를 구독하기 때문에, 외부 스레드를 사용하는 def로 라우터 함수를 선언하면 유저 수가 많아질 경우 그만큼 다른 api를 핸들링할 스레드 풀의 스레드 개수가 부족해지고 CPU 자원도 낭비되므로, async def로 선언하여 코루틴 사용
```
@api.get("", response_model=SSEMessage)
async def get_server_event(user_id: auth = Depends()):
    channel_name = RedisKey.get_s_queue_key(user_id)
    return EventSourceResponse(
        sse_service.subscribe_channel(channel_name)
    )
```

service

async def subscribe_channel(self, channel_name: str):
    try:
        q = RedisList(channel_name)
        while True:
            q_response = await q.get(is_blocking=True) # 큐에 데이터 쌓일 때까지 대기
            if q_response is not None:
                yield ServerSentEvent(
                    data=q_response,
                    retry=1000,
                )
                await asyncio.sleep(0.2)
    except asyncio.CancelledError as e:
        logger.info(f"{channel_name} -> {e}")
        return
    except BaseException as e:
        logger.error(f"{channel_name} -> {e}")
        return

message queue

특정 이벤트에서 put()을 통해 queue에 메세지가 전송되면 해당 queue를 구독하고 있는 코루틴에서 get()으로 그 메세지를 꺼내 클라이언트로 전송한다
클라이언트에 전달하는 메세지가 단순하고, ~~동시성 문제를 해결하기 용이한 Redis List 자료구조를 메세지 큐로 사용~~

class RedisList:
    """
    Redis Lists are an ordered list, First In First Out Queue
    Redis List pushing new elements on the head (on the left) of the list.
    The max length of a list is 4,294,967,295
    """

    def __init__(self, queue_name):
        self.key = queue_name
        self.client = config.redis

    def put(self, element):
        self.client.lpush(self.key, element)

    async def get(self, is_blocking=False, timeout=None):
        """sse router에 연결되어 redis 채널을 구독하는 해당 메서드만 비동기 처리"""

        if is_blocking:
            key, element = await config.aioredis.brpop(self.key, timeout=timeout)
        else:
            element = await config.aioredis.rpop(self.key)

        return element

HTTP Protocol Version + 백엔드 인프라 http 버전

SSE vs WS

트러블 슈팅

NGINX

<aside> 💡

개발 서버에 실행 중인 Nginx 때문에 SSE 연결이 정상 작동하지 않았다

</aside>

nginx 기본 설정

proxy_http_version 1.0;
proxy_set_header Connection close;
proxy_buffering on;

proxy_read_timeout 60s;

TOBE(예시)
```
proxy_http_version 1.1;
proxy_set_header Connection '';
proxy_buffering off;

# 필요한 경우 조절
proxy_read_timeout 1800s;
```
- HTTP 1.0 특성 상 지속 연결을 위한 헤더가 제공되지 않아 SSE 연결이 종료됨
- Connection 헤더 값이 close로 설정되면 SSE 연결이 종료됨
- proxy_buffering을 비활성화하여 서버의 응답을 버퍼링하지 않고 바로 전송
  - 활성화하게 되면 정상 작동하지 않거나 실시간성이 떨어진다 (전체 응답이 버퍼에 쌓이기 전까지 클라이언트에 보내지 않음)
  - 하지만 SSE 연결 서버가 분리되어있지 않은 경우 해당 WAS에 대한 nginx 설정에 버퍼링을 비활성화하면 SSE 연결을 제외한 API 응답에 대해서도 버퍼링을 하지 않으므로 비효율적일 수 있다
  - 따라서 기존 nginx 설정의 proxy_buffering은 건드리지 않고
    1. SSE 연결 엔드포인트에 대한 nginx 설정을 분리하고 그 분리된 설정에만 버퍼링 비활성화 처리를 하거나
    2. WAS에서 SSE 연결 API 응답 헤더에 X-Accel-Buffering: no를 포함하여 이를 수신한 nginx에서 SSE 관련 응답만 버퍼링하지 않도록 처리할 수 있다
- proxy_read_timeout 기본 설정은 60초 이내에 전달되는 데이터가 없는 경우 SSE 연결 종료

추가 고려 사항 및 TODO

대상그룹 HTTP 프로토콜 버전 업그레이드

데이터 유실

SSE 재연결

<aside> 💡

클라이언트에서 일정 시간 동안 이벤트를 수신하지 못하는 경우 타임아웃이 발생해 브라우저에서 서버에 재연결 요청을 보낸다

</aside>

서버에서 푸시할 때 항상 이벤트에 대한 id 값을 포함해야 한다
클라이언트는 SSE 재연결 이후 마지막으로 수신한 이벤트 ID를 Last-Event-Id 요청 헤더에 담아 보내면 서버는 마지막 이벤트 ID 이후의 데이터를 전부 푸시하는 로직이 필요하다