RxJS 빠르게 배우기 03 - Observer & Subscription & Operator

지난 글에서는 Observable과 구독 관련 동작 방식에 대해서 글을 작성했다. 이번 글에서는 Observer, Subscription, Operator에 대해서 간단하게 알아보자.

Observer

옵저버는 Observable에 의해 전달된 값을 소비하는 주체이다. 일종의 콜백 묶음으로 볼 수 있는데, Observable이 전달해주는 next, error, complete을 어떻게 처리하는지에 대한 내용이 있는 콜백들이다.

const observer = {
  next: x => console.log(`Next value: ${x}`),
  error: err => console.error(`Error: ${err}`),
  complete: () => console.log(`Complete`)
}

observable.subscribe(observer);

모든 경우를 처리해줄 필요 없이, 일부분만 설정해줘도 정상 동작한다. 다만 알람이 오게 될 때 해당 부분을 처리해주는 콜백이 없으면 무시하게 된다.

observable.subscribe(
  x => console.log('Observer got a next value: ' + x),
  err => console.error('Observer got an error: ' + err),
  () => console.log('Observer got a complete notification')
);

꼭 객체 형태로 전달해줄 필요도 없다. 위 예시처럼, next, error, complete 순서대로 인자값을 전달받게 된다.

Subscription

Subscription(구독)은 처리 가능한 자원을 나타내는 객체라고 설명하고 있다. 쉽게 보면 Observable을 실행하는 것이라고 볼 수 있다. Subscription은 unsubscribe라고 하는 중요한 메서드가 있는데, subscription에 묶여있는 자원을 해제하는 것이다.

설명을 줄줄 하면 더 헷갈리는 것 같은데, 쉽게 말해서 Observable을 구독하면 Subscription 객체를 리턴한다. 그 객체는 구독을 취소할 수 있는 unsubscribe 메서드를 가지고 있는 객체라고 볼 수 있다.

import { interval } from "rxjs";

const observable = interval(1000);

const subscription = observable.subscribe(console.log);

subscription.unsubscribe(); // 구독을 바로 취소한다.

Subscription은 여러 개를 함께 둘 수 있다. 그리고 한 번에 복수 개의 Subscription을 구독 취소할 수 있게 된다. 한 개로 묶는 방법은 add 메서드를 사용하면 된다. 반대로 함께 묶인 것을 제거하려면, remove(otherSubscription) 형태로 메서드를 사용하면 된다.

import { interval } from "rxjs";

const observable1 = interval(400);
const observable2 = interval(300);

const subscription = observable1.subscribe((v) => console.log(`first: ${v}`));
const childSubscription = observable2.subscribe((v) =>
  console.log(`second: ${v}`)
);

subscription.add(childSubscription); // Subscription을 하나로 묶어서 관리

setTimeout(() => {
  subscription.unsubscribe(); // 두 개 모두 한 번에 구독 취소 처리가 된다.
}, 1200);

// second: 0
// first: 0
// second: 1
// first: 1
// second: 2

Operator

RxJS는 보통 Operator(오퍼레이터)가 가장 유용하다. 복잡한 비동기 코드를 쉽고 명확한 방법으로 합성하게 해준다. 오퍼레이터는 함수이다. 두 가지 종류가 있는데, 하나는 Pipeable Operator이고 하나는 Creation Operator이다.

Pipeable Operator

Pipeable은 operatorInstance.pipe(operator())와 같은 문법을 사용하는 Observables을 통과할 수 있는 연산자 종류이다. 대표적으로는 filter(...), mergeMap(...) 등이 있다. 호출 되었을 때, 기존에 존재하는 Observable instance를 바꾸지는 않고, 새로운 Observable을 만들어낸다. 즉, 순수 함수의 특성을 가지고 있다.

Creation Operator

Creation Operator는 새로운 Observable을 만들기 위해 단일한 함수 호출이 가능한 함수이다. 예를 들자면, of(1, 2, 3)를 사용하면, 1, 2, 3 값을 내보내는 observable을 만들 수 있다. 예를 들자면 interval 함수는 숫자를 매개 변수로 받아서 Observable을 내보내는 함수이다.

import { interval } from "rxjs";

const observable = interval(1000 /* number of milliseconds */);

observable.subscribe(console.log);
// 0부터 1초마다 1씩 큰 수가 로그로 찍힌다.

위처럼 간단하게 특정하게 정해진 방식대로 독자적으로 Observable을 생성해주는 함수들을 Creation Operator라고 한다.

---

Operator를 사용한 예를 간단히 들어보자 map 오퍼레이터는 JS 배열의 메서드에서의 map 메서드와 유사한 역할을 한다.

import { of } from "rxjs";
import { map } from "rxjs/operators";

map<number, number>((x) => x * x)(of(1, 2, 3)).subscribe(console.log);

// 1
// 4
// 9

of(1, 2, 3)
  .pipe(map((x) => x * x))
  .subscribe(console.log);

// 1
// 4
// 9

first 오퍼레이터는 첫 번째 값을 새로운 Observable로 만들 수 있다.

import { first } from "rxjs";

first()(of(1, 2, 3)).subscribe(console.log);
// 1

of(1, 2, 3).pipe(first()).subscribe(console.log);
// 1

위 예시에서 pipe 메서드를 사용한 것을 공식 문서에서는 Piping(파이핑) 이라고 말 한다. 실제로 파이핑 하지 않고 작성한 코드는 쉽게읽히지 않게 되는 경우가 있다. 예를 들자면 op4()(op3()(op2()(op1()(obs)))) 이러한 느낌의 코드가 나올 수 있다. 이러한 이유로 옵저버블에서는 pipe 메서드를 제공하고 있다. 방금 예시의 읽기 힘든 코드는 아래와 같이 작성될 수 있게 된다.

obs.pipe(
  op1(),
  op2(),
  op3(),
  op3(),
)

공식 문서에 정말 많은 Operator를 설명해두었지만, 이번 글에서는 자세하게 다루지 않고, 디테일한 내용들은 다른 글에서 분리하여 정리해보려고 한다. 이 링크에서 오퍼레이터의 종류와 여러 오퍼레이터를 나열해두었다.

Reference

• RxJS 프로그래밍 (책)
• https://rxjs-dev.firebaseapp.com/guide/subscription