RxJS와 반응형 프로그래밍

Promise는 한 번 정착하는 결과 하나를 다루기에 알맞습니다. 그런데 사용자의 클릭이나 키 입력, 웹소켓의 메시지, 주기적인 타이머처럼 시간이 지나며 값이 여러 번 생기는 흐름은 하나의 Promise로 나타내기 어렵습니다. 반응형 프로그래밍(reactive programming)은 이런 흐름을 값의 스트림(stream)으로 보고, 값이 생길 때마다 무엇을 할지를 선언하는 방식입니다.

RxJS는 이 방식을 자바스크립트에서 쓰게 해 주는 라이브러리로, 스트림을 Observable이라는 객체로 나타내고 연산자로 바꾸고 거르고 합칩니다. 이 글은 Observable을 만들고 구독하는 과정과 구독을 끝낼 때의 정리, 연산자가 값을 처리하는 순서를 출력으로 따라가고, Promise와 비교해 도구를 고르는 기준을 정리합니다. 예제는 새 폴더에서 npm install [email protected]로 RxJS 7.8을 설치한 ES 모듈 프로젝트에서 확인합니다.

Observable과 구독

Observable은 구독(subscribe)한 쪽에 값을 밀어 보내는(push) 스트림입니다. 구독하는 쪽은 관찰자(observer)라 부르는 객체에 세 가지 함수를 담아 넘기는데, 새 값이 올 때 부르는 next, 오류로 끝날 때 부르는 error, 정상적으로 끝날 때 부르는 complete입니다.

import { Observable } from "rxjs";

const numbers = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => {
    subscriber.next(4);
    subscriber.complete();
  }, 100);
  return () => console.log("정리 함수 실행");
});

console.log("구독 전");
numbers.subscribe({
  next: (value) => console.log(`값: ${value}`),
  complete: () => console.log("완료"),
});
console.log("구독 후");

구독 전
값: 1
값: 2
값: 3
구독 후
값: 4
완료
정리 함수 실행

new Observable에 넘긴 함수는 정의할 때가 아니라 구독할 때 실행됩니다. 그래서 처음 세 값은 subscribe 호출 안에서 동기적으로 전달되어 구독 후보다 먼저 출력되고, 네 번째 값은 타이머가 끝난 뒤 전달됩니다. complete가 불리면 관찰자의 complete가 먼저 실행되고, 그다음 구독이 닫히면서 함수가 돌려준 정리 함수(teardown)가 실행됩니다.

이 Observable은 구독할 때마다 넘긴 함수를 새로 실행하므로, 두 곳에서 구독하면 값의 생산도 두 번 따로 일어납니다. 이렇게 구독마다 생산을 새로 시작하는 것을 차가운(cold) Observable이라 하며, 이미 진행 중인 생산을 여럿이 함께 나누어 받는 뜨거운(hot) 쪽은 뒤에서 Subject로 봅니다.

구독 해제와 정리

subscribe는 구독을 나타내는 Subscription 객체를 돌려주며, 이 객체의 unsubscribe()를 부르면 더는 값을 받지 않고 정리 함수가 실행됩니다. 다만 정리 함수가 무엇을 정리할지는 Observable을 만든 쪽이 정해야 합니다.

import { Observable } from "rxjs";

const cancellable = new Observable<number>((subscriber) => {
  subscriber.next(1);
  const timer = setTimeout(() => {
    console.log("타이머 본문 실행");
    subscriber.next(2);
    subscriber.complete();
  }, 1000);
  return () => {
    clearTimeout(timer);
    console.log("타이머 정리");
  };
});

const subscription = cancellable.subscribe({
  next: (value) => console.log("값:", value),
  complete: () => console.log("완료"),
});
setTimeout(() => subscription.unsubscribe(), 250);

값: 1
타이머 정리

250ms에 구독을 해제하면 정리 함수가 타이머를 취소하므로, 1초 뒤의 타이머 본문은 실행되지 않습니다. 구독 해제는 complete를 부르지 않으므로 완료도 출력되지 않습니다. 만약 정리 함수가 로그만 남기고 clearTimeout을 부르지 않았다면, 관찰자는 더 값을 받지 않더라도 예약된 타이머는 그대로 남아 1초 뒤에 타이머 본문 실행을 출력했을 것입니다.

pipe와 연산자

연산자(operator)는 Observable을 받아 새 Observable을 돌려주는 함수입니다. of나 interval처럼 Observable을 새로 만드는 생성 연산자와, pipe() 안에 차례로 늘어놓아 값을 바꾸고 거르는 연산자가 있으며, RxJS 7.2부터는 둘 다 "rxjs"에서 바로 가져올 수 있습니다.

import { interval, filter, map, take } from "rxjs";

interval(100)
  .pipe(
    take(5),
    filter((n) => n % 2 === 0),
    map((n) => `짝수: ${n}`),
  )
  .subscribe({
    next: (text) => console.log(text),
    complete: () => console.log("완료"),
  });

짝수: 0
짝수: 2
짝수: 4
완료

interval(100)은 0.1초마다 0부터 1씩 커지는 수를 내보내고, take(5)는 그 가운데 처음 다섯 개인 0부터 4까지만 통과시킨 뒤 스트림을 끝냅니다. filter는 그중 짝수인 0과 2, 4만 남기고, map은 그 세 값을 문자열로 바꿉니다. take의 개수는 그 연산자에 들어오는 값의 개수이므로, 최종 출력은 다섯 개가 아니라 세 개입니다.

위 그림은 이 파이프라인을 단계별로 따라갑니다. 원래 값 0은 filter를 통과해 출력되고, 1은 걸러지며, 2는 map을 거쳐 출력됩니다. 다섯 번째 값 4가 지나가면 take가 스트림을 끝내고, 그 끝냄이 관찰자의 complete로 전달된 뒤 interval의 타이머를 정리하는 정리 함수가 실행됩니다.

타입스크립트는 파이프라인의 각 단계에서 타입을 추론합니다. 위 예에서 interval은 Observable<number>이고, map을 지난 뒤에는 Observable<string>이 되어 next의 text가 문자열로 추론됩니다. 브라우저에서는 fromEvent(input, "input")로 입력 이벤트를 스트림으로 만들고, debounceTime(300)으로 입력이 0.3초 동안 멈추었을 때의 값만 받는 식으로 검색창을 만들 수 있습니다.

더 깊이: 값마다 시작하는 작업을 펼치는 네 연산자

검색창처럼 들어온 값마다 서버 요청 같은 비동기 작업을 새로 시작한다면, map으로는 Observable을 내보내는 Observable이 되어 버립니다. 이럴 때는 값마다 안쪽(inner) Observable을 만들고 그것을 구독해 나온 값을 바깥으로 내보내는 연산자를 씁니다. RxJS에는 이런 연산자가 넷 있으며, 앞의 안쪽 Observable이 아직 끝나지 않았는데 새 값이 들어왔을 때의 처리만 다릅니다. 아래 설명은 RxJS 7의 연산자 소스에 붙은 문서 주석을 따른 것입니다.

switchMap은 새 안쪽 Observable이 생기면 앞의 것에서 값을 내보내기를 멈추고 새것으로 바꾸며, 소스 코드에서도 앞의 안쪽 구독을 취소한 뒤 새로 구독합니다. 그래서 가장 최근 입력의 결과만 필요한 검색에 알맞습니다. mergeMap은 안쪽 Observable을 모두 함께 구독해 값이 나오는 대로 섞어 내보내고, 동시에 구독할 개수를 정하는 매개변수를 받습니다.

concatMap은 앞의 것이 끝나야 다음 것을 구독하는 방식으로, 동시 개수를 1로 둔 mergeMap과 같습니다. 문서 주석은 소스의 값이 안쪽 작업이 끝나는 속도보다 빠르게 끝없이 들어오면 기다리는 안쪽 Observable이 한없이 쌓여 메모리 문제가 생긴다고 경고합니다. exhaustMap은 반대로 앞의 것이 끝나지 않았으면 새 값을 무시하므로, 진행 중인 저장 요청이 끝나기 전에 다시 누른 버튼을 흘려보낼 때 쓸 수 있습니다.

Subject로 함께 받기

Subject는 직접 next를 불러 값을 내보낼 수 있는 Observable로, 구독한 모든 관찰자에게 같은 값을 나누어 보냅니다. 구독할 때 생산을 새로 시작하지 않으므로, 늦게 구독한 쪽은 그 전에 지나간 값을 받지 못합니다.

import { Subject } from "rxjs";

const subject = new Subject<number>();
subject.subscribe((value) => console.log("A:", value));
subject.next(1);
subject.subscribe((value) => console.log("B:", value));
subject.next(2);

A: 1
A: 2
B: 2

B는 값 1이 지나간 뒤에 구독했으므로 2만 받습니다. 늦게 구독해도 가장 최근 값을 받아야 한다면 BehaviorSubject를 쓰며, 이 차이는 이벤트나 상태를 여러 곳에서 함께 받을 때 중요합니다.

Promise로 바꾸기

Observable의 값을 await로 받아야 할 때는 firstValueFrom이나 lastValueFrom을 씁니다. 각각 첫 값이나 마지막 값으로 이행되는 Promise를 돌려주며, 예전의 toPromise는 RxJS 7에서 권장되지 않습니다.

import { interval, lastValueFrom, take } from "rxjs";

const last = await lastValueFrom(interval(100).pipe(take(3)));
console.log(last);

2

lastValueFrom은 스트림이 끝나야 이행되므로, take(3)처럼 끝나는 스트림에 써야 합니다. 끝나지 않는 interval(100)에 바로 쓰면 그 Promise는 영원히 정착하지 않고, 값 없이 끝난 스트림이라면 Promise는 거부됩니다. 이 예제는 파일의 최상위에서 await를 쓰므로 ES 모듈이어야 합니다.

도구 고르기

위 그림처럼 결과가 하나라면 Promise가, 소비하는 쪽이 하나씩 당겨 가는 흐름이라면 앞 글의 비동기 제너레이터나 다음 글의 비동기 이터러블이 더 단순합니다. 값이 생산하는 쪽에서 밀려오고, 여러 스트림을 합치거나 이전 요청을 버리고 새 요청으로 바꾸는 switchMap 같은 시간에 따른 제어가 필요할 때 Observable이 힘을 발휘합니다. 어느 도구를 쓰든 여러 구독자가 생산을 나눌지, 실패하면 다시 시도할지, 진행 중인 작업을 정말 취소할지는 따로 정해야 하며, Observable의 자세한 동작은 RxJS 7.8.2의 Observable 안내에 있습니다.

정리

• Observable은 구독한 관찰자에게 next로 여러 값을 밀어 보내고, error나 complete로 끝나며, 끝나거나 해제되면 정리 함수가 실행됩니다.
• 차가운 Observable은 구독할 때 생산을 시작하므로 구독마다 따로 실행되고, 처음 값들은 subscribe 호출 안에서 동기적으로 전달될 수 있습니다.
• 구독 해제는 complete를 부르지 않으며, 예약한 작업을 정말 멈추려면 정리 함수가 그 작업을 직접 취소해야 합니다.
• pipe 안의 연산자는 차례로 값을 처리하고, take의 개수는 그 연산자에 들어오는 값의 개수이며, 타입은 단계마다 추론됩니다.
• Subject는 값을 여럿에게 함께 보내 늦은 구독자는 지난 값을 받지 못하고, firstValueFrom과 lastValueFrom은 스트림을 Promise로 바꿉니다.

연습 문제

1. 연산자의 순서 바꾸기

interval(100).pipe(filter((n) => n % 2 === 0), take(5), map(...))처럼 filter를 take 앞으로 옮기면, 출력되는 숫자는 무엇이며 원래 값은 몇 개가 필요한가요?

2. 로그만 남기는 정리 함수

cancellable의 정리 함수에서 clearTimeout(timer);를 지우고 같은 순서로 실행하면 출력은 어떻게 달라지나요?

3. 끝나지 않는 스트림의 Promise

await firstValueFrom(interval(100))과 await lastValueFrom(interval(100))은 각각 어떻게 되나요?

다음 글 비동기 이터레이터와 for-await-of에서는 값을 하나씩 기다리며 당겨 오는 비동기 이터러블의 규칙과, 반복을 일찍 끝낼 때의 정리를 다룹니다.