[RxJS] 중첩 Observable을 처리하는 3가지 방법 - mergeMap(), concatMap(), switchMap()

이전까지는 여러 스트림의 출력을 동시에 하나로 결합하는 방법 살펴봄.
이번 내용에서는 옵저버블 자체에서 다른 옵저버블을 방출하는 “중첩 옵저버블" 처리하는 방법 살펴볼 예정.

 

---

1. Intro

(1) 중첩 옵저버블이란?

const search = Rx.Observable.fromEvent(inputText, 'keyup')
  //...
  .map(query => sendRequest(testData, query)) // sendRequest는 비동기작업하는 옵저버블
  // ...
  .subscribe(console.log)

• 사용자가 입력한 키워드를 포함하는 스트림을, 키워드에 대한 검색결과의 배열로 변환함.
• 근데 그 sendRequest 옵저버블 자체가 옵저버블 객체를 반환하는 상황.

중첩 옵저버블 개념도

 

// sendRequest()가 또다른 옵저버블 객체를 반환하는 경우 생성되는 값의 타입
Observable<Observable<Array>>

• 구독자는 중첩 옵저버블 값의 계층에 반응하여 Observable<Array>를 직접 처리해선 안됨.
• 이 중첩 옵저버블을 단일 계층으로 평탄화해야 함

 

(2) 데이터 평탄화란?

다차원 배열의 수준을 줄여야 하는 배열에서 개념이 비롯됨.

[[0, 1],[2, 3],[4, 5]].flatten(); // [0, 1, 2, 3, 4, 5]

[[0, 1],[2, 3],[4, 5]].reduce((a, b) => a.concat(b), []) // [0, 1, 2, 3, 4, 5]

• 중첩 옵저버블에서 값을 추출하고, 중첩 구조를 통합하여 사용자가 한 수준만 보게 함.
• flatten 하고 나서 결과 배열이 일차원이면 작업이 훨씬 쉬워짐.

---

2. mergeMap()

함수를 반환하는 옵저버블을 소스 옵저버블에 매핑하고 + 출력 옵저버블을 평탄화해서 반환

활용예시: 검색어 입력 후 결과 반환.

const search$ = Rx.Observable.fromEvent(searchBox, 'keyup')
  .pluck('target','value')
  .debounceTime(500)
  .filter(notEmpty)
  .do(term => console.log(`Searching with term ${term}`))
  .map(query => URL + query)
  .mergeMap(query => Rx.Observable.ajax(query) // 옵저버블 매핑 + 소스 옵저버블로 평탄화
		.pluck('response', 'query', 'search')
		.defaultIfEmpty([]))
	.map(R.map(R.prop('title'))) // 응답결과 배열의 모든 제목속성 추출
  .do(arr => count.innerHTML = `${arr.length} results`)
  .subscribe(arr => {
    clearResults(results);
    appendResults(arr, results);
  });

• fromEvent, pluck, map - 검색어를 입력 후 keyup 이벤트가 일어나면, 해당 검색어로 URL이 만들어짐
• mergeMap - 서비스 조회하는 옵저버블을 소스 옵저버블에 매핑 후 평탄화하고 있음
• map, subscribe - ajax로 서비스 조회한 결과를 구독함
• 비동기 호출 간의 의존성에 대한 걱정 없이, 키 입력이 검색 결과에 직접 매핑된 것처럼 키 입력에 관해 추론할 수 있음.

---

3. switchMap()

mergeMap과 비슷하지만, 최근에 매핑된 옵저버블 값만 표시함. (인터벌 2초라 하면 2초내에 응답 안오면 걍 다음걸로 기다림.)

활용예시: 주식시세 데이터 받아와서 표시하기

  const requestQuote$ = (symbols, fields) =>
    Rx.Observable
      .fromPromise(ajax(makeQuotesUrl(symbols,fields)))
      .pluck('quoteResponse','result');
  
  const twoSecond$ = Rx.Observable.interval(2000);

  const fetchDataInterval$ = symbol =>
    twoSecond$
     .switchMap(() => requestQuote$([symbol],['currency']))
     .map(extract);
  
  fetchDataInterval$('FB')
    .subscribe(logResult);

• requestQuote - 주식 심볼 이름으로 필요한 주식 시세 값을 가져오는 옵저버블
• twoSecond - 2초 인터벌 옵저버블
• fetchDataInterval - 키워드를 가지고, 2초 인터벌 옵저버블에 검색 결과를 요청하는 옵저버블을 매핑함.

 

(1) 모든 이벤트 마다 DOM을 불필요하게 갱신하고 있다면?

Rx.Observable.of("a", "b", "c", "d", "d", "e", "e", "e")
  .distinctUntilChanged()
  .subscribe(console.log); // a b c d e

distinctUntilChanged 개념

const fetchDataInterval$ = (symbol) =>
  twoSecond$
    .switchMap(() => requestQuote$([symbol], ["currency"]))
    .distinctUntilChanged(([symbol, price]) => price);

• 주가를 기준으로 비교를 수행하므로, 주가가 변경될 때만 DOM 갱신할 수 있음.

---

4. concatMap()

각 옵저버블은 이전 옵저버블이 완료될 때까지 대기함.

활용예시: 위젯 드래그 앤 드롭

const panel = document.querySelector('#dragTarget');
const mouseDown$ = Rx.Observable.fromEvent(panel, 'mousedown');
const mouseUp$ = Rx.Observable.fromEvent(document, 'mouseup');
const mouseMove$ = Rx.Observable.fromEvent(document, 'mousemove');

const drag$ = mouseDown$.concatMap(() => mouseMove$.takeUntil(mouseUp$));

drag$.forEach(event => {
  panel.style.left = event.clientX + 'px';
  panel.style.top = event.clientY + 'px';
});

• panel - 드래그하는 타깃
• mouseDown, mouseUp, mouseMove - 각 이벤트의 옵저버블
• drag - 마우스 이벤트를 방출하는 일련의 스트림.
• takeUntil - mouseUp 이벤트가 발생할 때까지, mouseDown과 연결된 이전 모든 mouseMove 이벤트를 가져옴.

concatAll 개념도

• 세 가지 마우스 이벤트 결합.
• 순서 유지 + 중첩 옵저버블 시퀀스 평탄화.
• takeUntil 사용하면, mouseUp이 값을 방출하는 즉히 mouseMove가 취소된다.

 

(1) concatMap() vs concatAll()

// 둘이 작동 방식이 동일함
const drag$ = mouseDown$.concatMap(() => mouseMove$.takeUntil(mouseUp$));
const drag$ = mouseDown$.map(() => mouseMove$.takeUntil(mouseUp$)).concatAll();

• concatMap()은 실제로는 map()...concatAll() 이다.

 

(2) map() + concat() ≠ concatMap()

const drag$ = mouseDown$.map(() => mouseMove$.takeUntil(mouseUp$)).concat();

• concat만으로는 중첩 옵저버블을 평탄화할 수 없음. 여러 옵저버블을 가져와서 순서대로 연결만 할 뿐, 고차 옵저버블끼리 작동하도록 설계되지 않음.