중첩 Observable 다루기: 고차 Observable을 평탄화(flattening)해서 값을 꺼내는 방법

1. 개념

고차 Observable(higher-order Observable)이란 각 값이 단순한 데이터가 아니라 또 다른 Observable인 스트림을 말한다. 예를 들어 fromEvent(button, 'click')을 구독하고, 클릭 이벤트마다 http.get(url) 같은 새로운 Observable을 반환하면, 바깥쪽 스트림(outer Observable)이 안쪽 스트림(inner Observable)을 값으로 방출하는 구조가 만들어진다.

이런 구조를 subscribe로 그냥 받으면 콜백 안에 inner$.subscribe(...)를 또 써야 하는 콜백 지옥이 생긴다. 이를 해결하기 위해 RxJS는 평탄화(flattening) 연산자를 제공한다. 평탄화란 Observable<Observable<T>> 형태의 중첩 구조를 Observable<T>로 펼쳐서 내부 값에 직접 접근할 수 있게 만드는 과정이다.

2. 왜 사용하는가

웹 개발에서 흔한 패턴은 "이벤트가 발생하면 비동기 작업을 시작한다"이다. 검색어 입력, 버튼 클릭, 라우트 변경 같은 이벤트 스트림에 반응해서 HTTP 요청이나 타이머 같은 또 다른 스트림을 시작해야 하는 경우가 대부분이다.

이때 단순히 map 연산자로 이벤트를 Observable로 변환하면 결과는 Observable<Observable<Response>>가 되어버려서, 실제 응답값을 꺼내려면 다시 구독해야 하는 이중 구독 문제가 생긴다. 평탄화 연산자는 이 변환과 구독, 값 전달을 한 번에 처리해서 코드를 선언적으로 유지시켜준다.

또한 각 평탄화 연산자는 "이전 내부 스트림을 어떻게 처리할 것인가"에 대한 전략을 내장하고 있어서, 요청 취소, 순서 보장, 중복 방지 같은 문제를 연산자 선택만으로 해결할 수 있다.

3. 동작 원리

평탄화 연산자는 공통적으로 다음 세 단계를 수행한다. 먼저 바깥 스트림에서 값이 들어오면 그 값을 이용해 내부 Observable을 생성하고, 그다음 내부 Observable을 구독해서 값을 바깥으로 흘려보내고, 마지막으로 여러 내부 Observable이 동시에 존재할 때 어떻게 처리할지 결정한다.

이 세 번째 단계에서 연산자별 차이가 발생한다. mergeMap은 모든 내부 Observable을 동시에 구독하고 값이 도착하는 대로 즉시 병합(merge)한다. 순서 보장이 없고 동시 실행이 가능하다.

concatMap은 이전 내부 Observable이 완료(complete)될 때까지 다음 내부 Observable의 구독을 미룬다. 큐(queue) 방식으로 순서를 엄격히 보장하지만 앞선 작업이 느리면 전체가 지연된다.

switchMap은 새로운 바깥 값이 들어오면 현재 구독 중인 내부 Observable을 즉시 구독 해지(unsubscribe)하고 새 내부 Observable로 전환(switch)한다. 오래된 요청을 취소하고 최신 요청만 유지하고 싶을 때 사용한다.

exhaustMap은 현재 내부 Observable이 진행 중이면 새로운 바깥 값을 무시(ignore)한다. 이미 실행 중인 작업이 끝나기 전까지 새 요청을 차단하고 싶을 때 사용한다.

4. 예제

검색창에 입력할 때마다 서버에 검색 요청을 보내는 상황을 가정한다. 먼저 fromEvent(inputEl, 'input')으로 입력 이벤트 스트림을 만들고, 그다음 pipe를 이용해 debounceTime(300)으로 입력이 멈춘 뒤 300ms를 기다린다.

이어서 switchMap(event => ajax.getJSON('/search?q=' + event.target.value))를 적용하면, 사용자가 계속 타이핑할 때마다 이전 요청은 취소되고 마지막 요청의 결과만 구독자에게 전달된다. 최종적으로 subscribe(result => console.log(result))로 결과를 출력하면 검색어가 바뀔 때마다 최신 응답만 화면에 반영된다.

반대로 파일 업로드처럼 순서가 중요한 작업이라면 concatMap을 사용한다. 파일 선택 이벤트 스트림에 concatMap(file => uploadFile(file))을 적용하면, 첫 번째 파일 업로드가 끝난 뒤에야 두 번째 파일 업로드가 시작되어 서버에 도착하는 순서가 보장된다.

버튼 연타를 방지하고 싶은 "저장" 버튼이라면 exhaustMap이 적합하다. fromEvent(saveBtn, 'click')에 exhaustMap(() => saveToServer(data))를 연결하면, 저장 요청이 진행 중일 때 버튼을 다시 눌러도 새 요청이 무시되어 중복 저장을 막을 수 있다.

5. 실무 사용 사례

자동완성(autocomplete) 검색 기능에서는 switchMap이 표준처럼 쓰인다. 사용자가 빠르게 타이핑할 때 이전 검색 요청 결과가 늦게 도착해서 최신 검색어 결과를 덮어쓰는 경쟁 상태(race condition)를 switchMap이 자동으로 방지한다.

순차적인 배치 작업이나 로그 전송처럼 순서가 데이터 무결성에 영향을 주는 경우에는 concatMap을 사용해서 요청이 도착한 순서대로 처리되도록 보장한다.

폼 제출, 결제 버튼, 좋아요 버튼처럼 중복 실행을 막아야 하는 UI 인터랙션에는 exhaustMap이 적합하다. 별도의 disabled 상태 관리 없이도 연산자 수준에서 중복 요청을 차단할 수 있다.

여러 개의 독립적인 알림 스트림을 하나로 합쳐서 실시간으로 모두 반영해야 하는 경우, 예를 들어 웹소켓 채널 여러 개를 동시에 구독할 때는 mergeMap을 사용해서 도착하는 순서와 무관하게 모든 이벤트를 즉시 흘려보낸다.

6. 주의할 점

mergeMap은 동시성 제어가 없기 때문에 짧은 시간에 많은 이벤트가 발생하면 동시에 수많은 내부 Observable이 구독되어 서버에 부담을 줄 수 있다. 필요하다면 mergeMap(fn, concurrent)처럼 두 번째 인자로 동시 실행 개수를 제한할 수 있다.

switchMap을 부주의하게 사용하면 완료되지 않은 부수 효과(side effect)가 취소 없이 계속 진행되는 경우가 있다. 예를 들어 서버에 이미 전달된 HTTP 요청은 클라이언트에서 구독을 해지해도 서버 쪽 처리는 취소되지 않을 수 있다는 점을 인지해야 한다.

concatMap은 내부 Observable이 완료되지 않으면 다음 값이 영원히 대기 상태에 머무른다. 완료되지 않는 스트림(예를 들어 인터벌 기반 스트림)을 concatMap에 넣으면 큐가 막혀버리는 문제가 생긴다.

exhaustMap은 진행 중인 작업이 있을 때 새 이벤트를 완전히 무시하기 때문에, 사용자에게 "요청이 무시되었다"는 피드백을 별도로 제공하지 않으면 혼란을 줄 수 있다.

7. 핵심 정리

중첩 Observable은 Observable이 값으로 다른 Observable을 방출하는 구조이며, 이를 그대로 두면 이중 구독 문제가 생긴다. mergeMap, concatMap, switchMap, exhaustMap은 모두 내부 Observable을 평탄화(flattening)한다는 공통 목적을 가지지만, 동시에 여러 내부 스트림을 어떻게 처리하는지에 따라 선택 기준이 달라진다. 최신 값만 필요하면 switchMap, 순서 보장이 필요하면 concatMap, 모든 값을 동시에 받아야 하면 mergeMap, 중복 실행을 막아야 하면 exhaustMap을 사용한다.