6. RxJs와 NgRx (1)

글쓴이는 리액티브X 공식 홈페이지와 RxJs 홈페이지를 많이 참고하였습니다.

1. Observable

옵저버블은 ES7 스펙으로 제안된 비동기 데이터 처리 표준으로 어떻게 동작하고 이점이 무엇인지 공부하였습니다.

 

일반적으로 동기적인 프로그래밍은 개발자가 작성한 코드가 한줄한줄 차례대로 실행되기를 기대하지만, 비동기 프로그래밍은 원하는 순간에 작성된 순서와 상관없이 실행하거나 보류됩니다.

 

비동기로 호출되는 이벤트를 옵저버블에 정의할 수 있으며 하나의 옵저버블에 다수의 비동기 이벤트로 이루어진 연속된 흐름(데이터 스트림)을 정의할 수도 있습니다.

 

즉 옵저버블 내부에 데이터를 처리하는 매커니즘이 있고, 옵저버블이 이벤트를 발생시키면 이 옵저버블을 관찰하고 있는 옵저버가 해당 이벤트를 감지하고 데이터를 받아 연산할 수 있습니다.

 

이러한 특징으로 인해 옵저버가 옵저버블을 구독한다고 말하며 동시성 연산이 가능하다고 합니다.

[공식문서 발췌] Obseravable이 배출하는 하나 또는 연속된 항목에 옵저버는 반응한다. 이러한 패턴은 동시성 연산을 가능하게 한다. 그 이유는 Observable이 객체를 배출할 때까지 기다릴 필요 없이 어떤 객체가 배출되면 그 시점을 감시하는 관찰자를 옵저버 안에 두고 그 관찰자를 통해 배출 알림을 받으면 되기 때문이다.

이를 다이어그램으로 표현하면 다음과 같습니다. (공식문서의 마블 다이어그램을 참고하였습니다.)

1.1 옵저버블 생성 및 실행

옵저버블이 무엇인지 알았으니, 이제 옵저버블을 생성하는 방법을 공부하였습니다.

 

다음은 RxJs에서 옵저버블을 생성하는 코드입니다.

 

 

 

 

 

 

 

 

먼저 Observable() 함수를 생성자 함수로 호출하여 첫번째 인자로, subscriber를 인자로 받아 처리하는 구독 함수를 생성합니다.

 

로직이 끝나고 이 옵저버블을 구독하는 모든 subscriber에게 이벤트를 방출(emit)하기 위해 subscriber.next() 메소드를 호출하여 결과 데이터를 전달합니다.

 

next()는 여러번 호출할 수 있으며 이벤트 방출이 완료되면 subscriber.complete() 메소드를 실행하여 방출을 종료합니다. complete() 메소드 이후의 next()로 방출되는 데이터는 전달되지 않으므로 방출을 종료할 시점에 호출해야 합니다.

 

실행 도중 에러를 catch하여 방출하고 싶을 때는 subscriber.error() 메소드를 호출하여 실패한 호출에 대응할 수 있습니다.

 

만약 옵저버블이 비정상적으로 종료되었다면 return 문에 종료 시 실행할 함수를 전달합니다.

(이부분은 마치 리액트의 hook에서 componentWillUnmount를 구현할 때 return문으로 함수를 전달하는 것과 똑같습니다.)

 

옵저버블 생성이 끝나면 옵저버는 필요한 시점에 해당 옵저버블을 구독하여 결과 데이터를 받을 수 있는데 바로 옵저버블의 subscribe() 메소드를 호출하면 됩니다.

 

subscribe() 메소드는 인자로 방출된 데이터를 인자로 하는 콜백함수를 정의할 수 있습니다. 단, subscribe의 콜백함수는 해당 옵저버블에 독립적이고 공유되지 않습니다.

[공식문서 발췌] This shows how subscribe calls are not shared among multiple Observers of the same Observable. When calling observable.subscribe with an Observer, the function subscribe in new Observable(function subscribe(subscriber) {...}) is run for that given subscriber. Each call to observable.subscribe triggers its own independent setup for that given subscriber.

 

다음 예제는 이전 장의 HttpClientModule 사용방법에서 나왔던 앵귤러 HttpClient.get 메소드를 옵저버블로 구현하고 있는 것을 직접 구현한 것입니다.

 

 

 

 

 

 

 

 

요청이 도중에 실패하였을 때 abort하기 위해 fetch api가 아닌 xmlHttpRequest를 활용하였습니다. (axios를 사용한 방법도 있습니다.)

1.2 Hot Observable vs Cold Observable

옵저버블은 hot/cold 두 종류가 있습니다.

Cold Observable

먼저 차가운(?) 옵저버블(cold observable)은 구독되기 전까지 데이터 스트림을 emit하지 않는 observable입니다.

 

옵저버가 옵저버블의 subscribe() 메소드를 호출하여 옵저버블을 구독하기 시작하면 그때서야 데이터 스트림의 처음부터 이벤트를 방출하는 옵저버블입니다.

 

옵저버가 어느 시점에 이 Cold Observable을 구독하는지와 상관없이 무조건 첫 이벤트부터 실행되기 때문에 각 옵저버에 옵저버블이 독립적으로 작동한다고 할 수 있습니다. 즉, 각 옵저버마다 독립적인 데이터 스트림이 방출됩니다. 이러한 동작 방식을 유니캐스트라고 합니다.

 

유니캐스트로 동작한다는 뜻은 곧 옵저버와 옵저버블이 1:1의 관계를 가진다고도 할 수 있습니다.

 

Cold Observable을 생성하는 것은 어렵지 않습니다. 기본적으로 옵저버블은 Cold Observable이기 때문입니다.

 

따라서 옵저버블을 생성하고 다수의 구독자가 어느 시점에 옵저버블을 호출하든 상관없이 데이터 스트림의 모든 이벤트가 방출됩니다.

Hot Observable

Hot Observable은 Cold와 반대로 옵저버블이 생성되자마자 이벤트를 방출하기 시작합니다.

 

옵저버가 구독하는 시점부터 옵저버블에서 방출되고 있는 이벤트를 받을 수 있습니다. 따라서 구독 이전에 방출된 데이터는 실행되지 않습니다.

 

Hot Observable을 만드는 방법으로 Subject를 사용하면 구현하기 편리합니다. Subject는 옵저버블이면서 옵저버이기 때문에 데이터를 구독하는 동시에 방출할 수도 있습니다.

 

Hot Observable을 구독하고 있는 옵저버는 부수효과(Side Effect)가 있기 때문에 멀티캐스트로 동작하는 특징이 있습니다. 즉, 모든 옵저버가 같은 데이터를 바라볼 수 있어 옵저버블과 옵저버가 1:M의 관계를 가집니다.

 

 

 

 

이 글에서 설명하는 Cold와 Hot Observable의 차이점 또한 흥미롭습니다.

 

글에 따르면 Cold Observable은 producer*가 내부에 생성되고 동작하며, Hot Observable은 producer가 바깥에 생성되어 있고 이를 참조하여 동작하는 차이점이 있습니다.

* producer: 옵저버블이 시작되는 지점을 일컫습니다. DOM 이벤트가 될수도 있고, 웹 소켓, 배열 등 observer.next()의 인자로 전달할 값이 곧 producer입니다.

 

글의 예시로 웹 소켓의 생성 위치에 따라 옵저버블이 cold인지 hot인지 구분할 수 있습니다.

// This is Cold Observable
const source = new Observable((observer) => {
  const socket = new WebSocket('ws://someurl');
  socket.addEventListener('message', (e) => observer.next(e));
  return () => socket.close();
});

// This is Hot Observable
const socket = new WebSocket('ws://someurl');

const source = new Observable((observer) => {
  socket.addEventListener('message', (e) => observer.next(e));
});

위의 예제처럼 옵저버블을 구독할 때마다 웹 소켓 연결 인스턴스를 생성하는 것은 매우 낭비적이므로 옵저버블 바깥 스코프에서 생성 후 이를 참조하는 것이 더욱 효율적입니다. 따라서 Hot Observable을 사용하는 것입니다.

2. RxJs 기본 사용법

RxJs는 ReactiveX 프로젝트에서 시작된 Reactive Programming을 지원하는 자바스크립트 라이브러리입니다. 언어에 따라 RxJava, RxSwift 등 다양한 라이브러리가 존재합니다.

2.1 Reactive Programming

본격적으로 RxJs를 시작하기 전에 RxJs가 지원하는 리액티브 프로그래밍이 무엇인지 알고 가는 것이 중요합니다.

 

리액티브 프로그래밍은 동기이냐 비동기이냐에 상관없이 데이터를 생산하는 그 어떤 것이든 연속적으로 흐르는 데이터 스트림 상에서 처리한 후 데이터 스트림을 구독하여 상태 변화에 반응하는 방식으로 동작하는 어플리케이션을 작성하는 프로그래밍 패러다임입니다.

 

장황한 설명처럼 보이지만 밑줄친 특징은 옵저버블의 특징과 일맥상통하다는 것을 알 수 있습니다. 따라서 RxJs는 어떤 데이터든 처리할 수 있고 연속성을 갖는 데이터를 처리하는 데에 적합한 옵저버블을 사용하여 데이터 스트림을 생성하고 방출합니다.

 

리액티브 프로그래밍은 옵저버 패턴, 이터레이터 패턴 그리고 함수형 프로그래밍 패러다임을 적절하게 활용하는 프로그래밍으로 알려져 있습니다.

 

각 패턴의 특징을 알면 리액티브 프로그래밍을 학습하는 데 도움이 되므로 간략하게 소개하겠습니다.

• 옵저버 패턴 : 객체의 상태 변화를 관찰하는 옵저버들의 목록을 객체에 등록하고 상태가 변할 때마다 함수(메소드)를 통해 객체가 직접 옵저버 목록에 존재하는 모든 옵저버들에게 이를 알리는(notifying) 디자인 패턴
• 이터레이터 패턴 : 구체적인 컬렉션 구현 방법을 외부로 노출하지 않으면서 그 집합체(컬렉션) 내부의 모든 항목에 접근할 수 있도록 방법을 제공하는 패턴. 이터레이터 인터페이스를 구현하여 반복적인 로직을 관리하는 패턴
• 함수형 프로그래밍 : 외부 데이터를 참조하지 않는 순수함수들을 조합하여 기능을 구현하는 프로그래밍

너무 간단하게 소개했지만 개인적으로는 이정도 개념만 알고 있어도 리액티브 프로그래밍을 지원하는 RxJs를 사용하는 데 큰 무리가 없었습니다.

물론 더 공부하고 찾아보면 좋습니다 :)

2.2 RxJs 사용하기

RxJs는 다음 순서로 작성합니다.

 

1. 데이터 소스를 Observable로 정의합니다.
2. rxjs에서 제공하는 operators로 데이터를 변경 및 추출합니다. 혹은 여러 개의 Observable을 하나의 Observable로 합치거나 분리할 수 있습니다
   
3. 원하는 데이터를 받아 처리하는 Observer를 정의합니다.
   
4. Observable.subscribe()를 통해 Observer를 등록합니다.
   
5. 데이터 사용이 완료되면 Observable의 구독을 정지하고 자원을 해제합니다.

 

rxjs는 옵저버블을 손쉽게 생성할 수 있는 다양한 함수를 제공합니다.

대표적인 키워드 세 가지를 보겠습니다. 자세한 명세 및 더 다양한 함수는 공식문서의 REFERENCE를 참고하십시오.

fromEvent(target: EventTarget, event: string) : target DOM엘리먼트의 이벤트를 데이터 소스로 하는 옵저버블을 반환
of<T>(args: T) : args를 옵저버블로 반환
from<T>(args: T) : args는 이터러블, Promise 혹은 배열(유사배열 포함)이며, 내부 값들을 하나씩 next()메소드에 반복 전달합니다.

다음은 a태그 클릭 이벤트를 로깅하는 예제입니다.

 

 

 

 

rxjs는 다양한 연산자를 제공하여 subscribe()에 도달하기 전에 옵저버블의 생성, 변환, 필터링, 에러 처리 기능을 더욱 쉽게 작성할 수 있도록 합니다. 연산자에는 Creation/Pipable 두 종류가 있습니다.

[참고] 사실 공식문서에서 나누는 기준은 더욱 세세하지만 이를 전부 구분하기가 어려워 큰 특징을 가지는 연산자 종류로만 구분지었습니다. 자세한 건 여기를 참고하십시오.

 

Creation 연산자는 말 그대로 독립적으로 실행되어 옵저버블을 생성하는 연산자로 위의 of, from, fromEvent도 이에 해당합니다. 이외에도 map, filter 등 마치 배열을 가공하는 것과 같은 함수 또한 제공합니다.

 

Pipable 연산자는 옵저버블의 pipe() 메소드의 콜백함수로 사용하여 기존의 옵저버블을 변경하지 않고 새 옵저버블을 생성하는 순수함수입니다.

 

여기서 Observable.pipe() 메소드는 기존의 옵저버블을 input으로 삼아 새 옵저버블을 생성하는 순수함수입니다. 내부에 여러 개의 콜백 함수를 두어 순차적으로 실행할 수 있습니다.

 

아래 예제는 pipe()의 한 예시로 주석으로 그 의미를 파악할 수 있도록 하였습니다.

fromEvent(btn, 'click')
.pipe(
  take(1), // 만약 여러개의 옵저버블 이벤트가 실핼되면 하나의 이벤트만 방출하겠다
  tap((_) => this.setLoading(true)), // tap은 별도의 로직을 작성하기 위해 쓰이는 연산자로, 여기서는 setLoading()을 호출
  concatMap(() => // high order mapping operator중 하나로 inner observable의 순서를 보장
    this.saveOrder().pipe( // 옵저버블을 반환하는 함수 saveOrder() 호출
      timeout(5000), // 시간 제한 5초
      catchError((e) => of({ status: '9000' })) // 5초 지나도 정상 응답이 안오면 {status: '9000'} 반환
    )
  )
)
.subscribe(() => {})

A Pipeable Operator is a function that takes an Observable as its input and returns another Observable. It is a pure operation: the previous Observable stays unmodified.

rxjs에서 제공하는 연산자가 많기 때문에 이를 한번에 외우는 게 어려워 rxjs 기반 실습을 하면서 어떤 오퍼레이터가 있는지 공부하였습니다. 실습에서 어떤 오퍼레이터가 있는지 소개하겠습니다.

 

하지만 그 중 High Order Mapping Operator는 중첩된 Observable을 이해하는 데 매우 중요한 개념이므로 미리 공부하겠습니다.

3. High Order Mapping Operator의 종류와 용도

[출처] What Is a Higher-Order Observable?

[출처] Comprehensive Guide to Higher-Order RxJs Mapping Operators

 

옵저버블로부터 방출된 데이터로 또다른 옵저버블을 방출하는 경우, inner observable을 방출하는 outer observable을 high-order observable이라고 합니다. 아까 pipe() 메소드를 설명하면서 기존의 옵저버블을 인자로 하여 새로운 옵저버블을 생성한다고 한 것처럼 하나의 옵저버블을 여러 개의 옵저버블로 분리할 수 있습니다.

 

정확한 쓰임새를 이해하기 위해 이 글에서 소개하는 예시를 참고하였습니다.

예시는 productId로 해당 product를 공급하는 공급자 정보를 http 요청으로 가져오는 상황입니다.

this.http.get(`/api/products/${productId}`)
.pipe(
  tap(res => res.supplierIds.map(
    supplierId => this.http.get(`/api/suppliers/${supplierId}`)
    		.subscribe(console.log) // inner observable subscribe
  )
).subscribe(console.log) // outer observable subscribe

옵저버블은 구독해야만 데이터를 방출하기 때문에 내부 옵저버블 또한 구독해야 합니다.

 

지금은 한 번만 중첩되었지만 만약에 옵저버블이 n번 중첩되었다면 각각에 대한 구독을 신경써야 하는 것은 여간 힘든 일이 아닐 수 없습니다. 또한 더 깊게 중첩되면 중첩될수록 언제 inner observable의 구독을 해지해야 하는지 그 시점을 파악하기 어려워지고 결국 옵저버블의 선언적인(declarative) 특징과 선언형의 장점을 무색하게 만듭니다.

 

RxJs에서는 이러한 High Order Observable의 불편한 점을 해결해주고자 High Order Mapping Operator를 제공하여 중첩된 구독(Nested Subscription)을 피하고 inner observable의 subscribe/unsubscribe를 쉽게 해줍니다.

 

그 대표적인 종류로 concatMap, switchMap, exhaustMap을 알아보겠습니다.

소개하는 것 이외에도 더 있지만 실무에서 주로 쓰이는 연산자들만을 소개하겠습니다.

 

연산자의 동작을 이해하기 위해 위의 공급자 예제를 다이어그램으로 표현하여 흐름을 파악해봅니다.

3.1 concatMap

concatMap은 'concatenate'의 뜻에 맞게 연속된 옵저버블의 순서를 보장합니다. 각각의 inner observable이 완료될 때까지 대기하고 옵저버블의 이벤트 방출이 완료되면 다음 inner observable로 진행하는데 각 결과를 이전 옵저버블 결과에 연결(concatenate)합니다.

 

concatMap은 모든 inner observable의 순서를 보장해야 할 때 사용하면 유용한데 예를 들어 form 내 모든 값이 정상적인 흐름으로 서버에 저장되었음을 보장할 때 사용할 수 있습니다.

Use concatMap if you want to ensure that each inner Observable is processed one at a time and in order. This is a great technique to use when updating or deleting data to ensure each operation is processed in sequence.

this.http.get(`/api/products/${productId}`)
.pipe(
  concatMap(res =>
    from(rse.supplierIds)
    .pipe(
      map(supplierId => this.http.get(`/api/suppliers/${supplierId}`)
    )
  )
).subscribe(console.log)

concatMap 예제

위의 공급자 예제는 concatMap을 적용하면 아래의 순서로 동작할 것입니다.

• P1이 방출되면 concatMap은 첫번째 inner observable을 구독
• S1이 방출될 때까지 기다리고 방출되면 결과 스트림에 결과값을 추가
• P2도 위의 과정을 동일하게 반복
• 모든 inner observable이 완료되면 결과 스트림 종료

3.2 switchMap

switchMap은 새로운 inner observable을 구독하기 시작하면 이전의 구독 중이던 inner observable의 구독을 해제하고 새 inner observable로 스위칭합니다.

 

대표적인 활용 예시로 검색창에 검색어가 매번 입력될 때마다 검색 이벤트를 방출하는 것은 매우 비효율적이므로 특정 시간동안 입력 이벤트가 발생하지않으면 그때서야 비로소 검색을 수행하는 디바운싱을 적용할 때 유용합니다.

this.http.get(`/api/products/${productId}`)
.pipe(
  switchMap(res =>
    from(rse.supplierIds)
    .pipe(
      map(supplierId => this.http.get(`/api/suppliers/${supplierId}`)
    )
  )
).subscribe(console.log)

switchMap 예제

switchMap의 동작 순서는 다음과 같습니다.

• P1이 방출되면 switchMap이 S1을 구독
• P2가 방출되면 S1의 구독을 해제하고 S2를 구독
• 모든 과정이 완료되면 결과 스트림은 가장 마지막 결과값만을 보유

3.3 exhaustMap

exhaustMap은 switchMap과 반대로 inner observable이 진행되는 동안에 방출된 새로운 inner observable은 구독하지 않습니다. 반드시 진행 중인 inner observable이 완료되어야 새로 방출된 inner observable을 구독할 수 있습니다.

 

활용 사례로 사용자가 저장 버튼을 연속적으로 빠르게 클릭하여 불필요한 서버 트래픽이 발생하는 것을 방지할 수 있습니다.

this.http.get(`/api/products/${productId}`)
.pipe(
  exhaustMap(res =>
    from(rse.supplierIds)
    .pipe(
      map(supplierId => this.http.get(`/api/suppliers/${supplierId}`)
    )
  )
).subscribe(console.log)

exhaustMap 예제

exhaustMap의 동작 순서는 다음입니다.

• P1이 방출되고 S1을 구독
• S1 구독 도중에 P2가 방출
• exhaustMap에 의해 S2 inner observable이 생성되지 않음
• 모든 과정이 완료되면 결과 스트림에는 S1만 존재

4. 메모리 누수

앵귤러에서 rxjs의 옵저버블 기반 데이터 스트림 처리 시 메모리 누수가 발생할 수 있습니다.

 

바로 옵저버가 옵저버블을 구독하고 모든 작업이 완료되었거나 옵저버블을 구독한 컴포넌트가 더이상 사용되지 않음에도 불구하고 구독을 해지하지 않았을 때입니다.

 

따라서 반드시 옵저버블이 더이상 필요하지 않을 때에는 옵저버블 구독을 해제해야 하는데 앵귤러에서는 해제할 수 있는 방법을 아래처럼 세가지를 제공합니다.

• ngOnDestroy 라이프사이클 훅에서 컴포넌트가 destroy되면 unsubscribe
• async pipe 사용
• take, takeUntil, takeWhile과 같은 필터링 연산자를 사용하여 특정 조건을 만족할 때까지 observable에 대한 구독을 관리

특히 async 파이프는 HTML 템플릿에서 옵저버블 데이터에 사용하여 옵저버블을 자동으로 구독하고 컴포넌트가 destroy되면 자동으로 unsubscribe해주는 아주 유용한 방법입니다.

5. Subject

Hot Observable을 만드는 방법에서 옵저버블이자 옵저버인 Subject에 대해 잠깐 언급했었습니다. 이제 Subject의 특징과 종류 및 용도에 대해 공부하였습니다.

 

옵저버블은 기본적으로 옵저버가 구독하기 전까진 데이터를 방출하지 않지만 서브젝트는 생성과 동시에 이벤트를 방출하기 시작하며, 모든 옵저버에게 동일한 코드를 실행하는 멀티 캐스트 방식으로 동작합니다.

 

또한 서브젝트가 어떻게 구현되어 있는지 내부 구현코드를 살펴보면 알 수 있듯이 내부적으로 subscribe() 메소드는 값을 전달하는 새로운 실행을 호출하지 않습니다. 마치 이벤트 리스너처럼 단순히 옵저버를 observers 리스트에 추가할 뿐입니다.

 

그리고 next() 메소드 또한 각 옵저버의 next()에 value를 전달해서 반복적으로 호출하는 것 뿐입니다.

어려울 줄 알았지만 의외로 단순하게 동작합니다 :)

/** In rxjs/src/internal/Subject.ts */

observers: Observer<T>[] = [];

next(value: T) {
    if (this.closed) {
      throw new ObjectUnsubscribedError();
    }
    if (!this.isStopped) {
      const { observers } = this;
      const len = observers.length;
      const copy = observers.slice();
      for (let i = 0; i < len; i++) {
        copy[i].next(value!);
      }
    }
}

/** @deprecated This is an internal implementation detail, do not use. */
  _subscribe(subscriber: Subscriber<T>): Subscription {
    if (this.closed) {
      throw new ObjectUnsubscribedError();
    } else if (this.hasError) {
      subscriber.error(this.thrownError);
      return Subscription.EMPTY;
    } else if (this.isStopped) {
      subscriber.complete();
      return Subscription.EMPTY;
    } else {
      this.observers.push(subscriber); // 
      return new SubjectSubscription(this, subscriber);
    }
 }

다음은 옵저버블과 서브젝트의 특징과 언제 활용하면 좋을지 정리한 표입니다.

Observable	Subject
각 옵저버에 유니캐스트로 동작하여 구독 중인 옵저버들 간 데이터 공유가 없음	모든 옵저버에 같은 코드를 실행하여 옵저버들 간 데이터를 공유함
옵저버블만 생성해서 데이터를 생성하고 방출	옵저버블과 옵저버 두 역할을 모두 할 수 있음
각 옵저버에만 특정된 데이터를 방출해야 할 때	모든 옵저버가 동일한 데이터를 바라보고 있어야 하고, 데이터의 저장과 수정이 빈번할 때

5.1 Subject의 종류

subject에는 몇가지 특화 subject가 존재하는데 그 종류와 특징에 대해 알아보았습니다.

• BehaviorSubject : 구독이 되는 시점부터 가장 최근에 방출된 데이터를 방출하며 그 이후 새로운 옵저버가 구독을 시작할 때 이전에 방출한 최근 값을 전달할 수 있습니다.
  첫번째 인자로 초기값을 가질 수 있습니다. 따라서 async 파이프 사용 시 컴포넌트 초기값을 지정하기 애매한 경우가 있는데 Subject는 초기값이 없어 문제가 되지만 BehaviorSubject는 초기값이 있어 문제가 되지 않습니다.
• ReplaySubject : 옵저버의 구독 시점과 관계없이 모든 데이터를 방출합니다. 첫번째 인자로 n개만큼 버퍼를 설정하면 가장 마지막 옵저버블부터 n개까지의 데이터를 방출합니다.
• AsyncSubject : 옵저버블로부터 가장 마지막에 방출된 값만을 방출합니다. 주의할 점은 옵저버블의 동작이 완료되고 결과값을 방출해야만 값을 방출하므로 반드시 complete()를 호출해야 합니다.

공식문서의 Subject 설명에서 동작 방식을 마블 다이어그램으로 잘 묘사하였으니 확인하면 유용합니다.

 

실무에서 특히 BehaviorSubject를 주로 활용하여 가장 최근에 방출된 값을 보장하는데 쓰였습니다.

 

예를 들어 지도 api가 정상적으로 완료되었는지에 따라 true/false를 next()로 값을 주입하여 구독시점에 지도가 정상적으로 떴는지 체크하는 로직을 확인할 수 있었습니다.

6. [실습] 검색 기능 구현

지금까지 만들었던 컴포넌트에서 이제 군데군데 rxjs를 활용해보겠습니다.

 

첫번째로 메뉴 컴포넌트에 검색 입력란을 두고 검색 기능을 구현하겠습니다. 검색어가 한 자 한 자 입력될 때마다 검색 로직을 호출하는 것은 매우 비효율적입니다. 따라서 디바운싱으로 300ms를 주어 실행하겠습니다.

 

먼저 전체적인 메뉴 모듈의 구조를 변경할 예정입니다. 예전에는 아래 코드처럼 컴포넌트에서 바로 임시데이터인 products를 불러와서 사용했었습니다.

 

 

그러나 검색 컴포넌트와 홈 컴포넌트 두 개가 동시에 사용할 데이터이기 때문에 이를 서비스로 분리하여 정의하는 편이 이젠 좋을 거 같으므로 메뉴 서비스로 분리하였습니다.

 

 

메뉴 서비스에서 검색 로직을 담당하는 doSearchResult 메소드는 옵저버블을 반환하고, 해당 옵저버블을 자세히 보면 다음과 같이 해석할 수 있습니다.

doSearchResult(value: string): Observable<Product.Coffee[]> {
  return isEmpty(value) // value가 빈 값이 아니면
  ? of(this.getProducts()) // 전체 products를 옵저버블로 리턴
  : of(this.products$).pipe( // 전체 products 중
    map((products: Product.Coffee[]) =>  
      products.filter((coffee: Product.Coffee) => coffee.title.includes(value)) // value가 product의 title에 포함되어 있는 product만 매핑
    )
  );
}

이제 MenuService를 검색 컴포넌트와 홈 컴포넌트에 DI한 후 products 데이터를 불러올것입니다.

먼저 홈 컴포넌트는 단순히 전체 products 데이터를 가져오면 되므로 ngOnInit에서 아래와 같이 정의합니다.

 

 

이제 검색 컴포넌트를 생성하여 옵저버블을 리턴하는 검색 로직을 활용할 것입니다.

 

가장 먼저 input 엘리먼트를 생성 후 input 엘리먼트를 참조하는 템플릿 변수를 정의합니다.

[참고] Reactive Form에서는 이를 formControl로 정의하여 valuechanges 옵저버블을 활용할 수 있습니다. 이는 마지막 장에서 활용해보도록 하겠습니다.

 

 

@ViewChild()에서 input 템플릿 변수를 참조하는 reference를 정의하고 fromEvent 연산자를 활용하여 입력 이벤트가 발생할 때마다 동작하는 옵저버블을 구현할 것입니다.

 

debounceTime 연산자는 인자로 주어진 number(ms)후에 이벤트를 방출하는 연산자입니다. 이를 활용하면 300ms만큼 디바운싱을 적용할 수 있습니다.

 

또한 deSearchResult() 메소드는 fromEvent 옵저버블의 inner observable이므로 적절한 High Order Mapping Operator를 적용해야 할 것입니다.

 

이미 공부할 때 스포했지만 switchMap은 이전에 진행중이던 inner observable을 abort하고 새로 발생한 inner observable을 구독하므로 디바운싱에 적합한 연산자입니다. 따라서 switchMap 연산자에서 doSearchResult()를 실행하겠습니다.

 

마지막으로 검색 결과를 자식 컴포넌트인 검색 컴포넌트에서 부모 컴포넌트인 홈 컴포넌트에게 데이터를 전달하기 위해 @Output()으로 이벤트를 전달합니다.

@Output() 사용방법은 이젠 따로 언급은 안할 예정이지만 부모 컴포넌트에서 해당 이벤트를 전달해야합니다. 1. Component와 Directive (1)을 참고해주세요 :)

 

검색 컴포넌트의 최종 코드는 다음과 같습니다.

 

 

여기까지 완료되었다면 아래처럼 검색 결과에 맞춰 메뉴가 보일 것입니다.

(좌) 빈 검색창, (우) '아메' 입력하고 300ms 후

참고로 검색한 결과값이 없다면 결과값이 없음을 표현하는 것 또한 좋은 ux입니다:)

<div *ngIf="products.length > 0;else empty">
  <!-- 중략 -->
</div>
<ng-template #empty>
  <strong>검색 결과가 없습니다.</strong>
</ng-template>

다음 글에서 NgRx를 활용하여 상태 관리를 하는 방법에 대해 알아보겠습니다.