2 posts tagged with "rxjs"

개요​

• 앵귤러 같기도 하고 코드를 머리를 써서 읽어야한다는 것 때문에 최대한 안 쓰고 싶다.
• 하지만 async/await 로 동시성 제어가 힘드므로 이 쪽이 답인 것 같다.

소스​

• 데이터 배열에서 5개씩 끊어서 병렬 요청한다.
• delay, catchError, delayWhen 의 조건만 다르게 하여 사용하면 된다.

import { EMPTY, from, of } from "rxjs";
import {
  catchError,
  delay,
  delayWhen,
  finalize,
  map,
  mergeMap,
  retry,
  tap,
  toArray,
} from "rxjs/operators";

// 슈도코드임
const fetchObservable = (data) => {
  return from(
    new Promise((resolve) => {
      setTimeout(() => {
        resolve({
          data,
        });
      }, 300);
    }),
  );
};

const concurrency = 5;
const fetchConcurrently$ = from(YOUR_DATA).pipe(
  mergeMap((token) => {
    return fetchObservable(YOUR_DATA).pipe(
      map(({ data }) => data),
      delay(1000),
      retry(1),
      catchError(() => EMPTY),
    );
  }, concurrency),
  map((data) => {
    return data.id;
  }),
  toArray(),
  delayWhen(() => Promise.resolve()),
  finalize(() => console.log("done")),
);

fetchConcurrently$.subscribe((ids) => console.log(ids));

테스트​

• 테스트 시에 jest 에서 done callback 을 적절히 호출해주자.

import { TestScheduler } from "rxjs/testing";
import { mergeMap, map, toArray } from "rxjs/operators";
import { from } from "rxjs";

describe("fetchConcurrently$ observable", () => {
  let testScheduler;

  beforeEach(() => {
    testScheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });
  });

  it("should produce expected output", () => {
    testScheduler.run((helpers) => {
      const { cold, expectObservable } = helpers;

      // Mock fetchObservable function
      const fetchObservable = (data) => {
        return from([data]);
      };

      // Mock YOUR_DATA
      const YOUR_DATA = [1, 2, 3, 4, 5];

      const concurrency = 5;
      const fetchConcurrently$ = from(YOUR_DATA).pipe(
        mergeMap((token) => {
          return fetchObservable(token).pipe(
            map((data) => data),
            // ... other operators
          );
        }, concurrency),
        map((data) => data),
        toArray(),
      );

      const expectedMarble = "(abcde|)";
      const expectedValues = { a: 1, b: 2, c: 3, d: 4, e: 5 };

      expectObservable(fetchConcurrently$).toBe(expectedMarble, expectedValues);
    });
  });
});

여담​

• marble 테스트도 작성해보고 싶었는데 레퍼런스로 삼을만한 문서를 못 찾아서 아쉽다.
  • GPT4가 위처럼 잘 짜줬다.
• 매번 of 와 from 을 헷갈리는데, 전자는 하나씩이고 후자는 덩어리다.

RxJS​

리액티브 프로그래밍은 개인적으로 비동기 프로그래밍과 함수형 프로그래밍의 종착지라고 생각한다. Stage1 Draft로 제안되어 더 이상 피할 수 없는 Observable 을 알아보자.

정의​

위키피디아에선 다음과 같이 정의되어있다.

reactive programming is a declarative programming paradigm concerned with data streams and the propagation of change. 데이터 스트림과 변화의 전파에 중점을 둔 프로그래밍 패러다임

2011년 태초에 RX 의 개념을 만든 MS 문서에서는 Reactive eXtensions 다음과 같다.

Reactive Extensions (Rx) is a library for composing asynchronous and event-based programs using observable sequences and LINQ-style query operators. 옵져버블 시퀀스와 링큐 쿼리 연산자를 사용하는 비동기, 이벤트 기반 프로그래밍 라이브러리

조금 더 디테일하게 말하면 데이터스트림을 Pulling 방식의 이터레이터 패턴인 IEnumerable<T>/IEnumerator<T> 로 만들어 Pushing 방식의 옵저버 패턴인 IObservable<T>/IObserver<T>로 전파/구독하는 것이다.

Rx를 모두 이해한 뒤 이 정의를 보면 어쩜 이렇게 깔끔하게 한 줄로 이 내용을 다 담았을까? 란 생각이 드는데, 처음보는 입장에선 비동기인 건 알겠네 정도로만 이해가 되는 듯하다. 어려운 게 당연하다. 멀티쓰레드 프로그래밍을 처음 배울 때의 감정을 생각해보자.

용어​

• 비동기 프로그래밍: 이 문서를 볼 정도면 설명이 필요 없을 것 같다.
• 함수형 프로그래밍: 함수를 FirstClass 로 취급해 파라미터, 리턴, 변수에 할당 가능하며 함수 합성이 가능하고 원하는 시점에 호출이 쉽다.
• 데이터스트림: file stream, event stream, http stream 의 친구로 data 식 표현
• 이터레이터: 설명 필요 없을 듯.
• 옵져버: 디자인패턴 책의 옵져버패턴 단원 참조, 자바스크립트에선 이미 이벤트리스너가 옵져버 패턴이다.
• 옵져버블: 특정 객체를 관찰하는 옵저버에게 여러 이벤트나 값을 전파하는 역할
• 풀: 데이터를 받을지 결정
• 푸쉬: 데이터를 보낼지 결정
• 싱글: 하나의 값이나 이벤트를 다룸
• 멀티플: 여러 개의 값이나 이벤트를 다룸

이 모든 용어를 하나의 표로 정리하면 다음과 같다.

패턴​

• 옵져버블의 변수 스타일은 이름 뒤에 const click$ 처럼 $을 뒤에 붙혀주는게 정형화되어있다.
• subscribe 는 next, error, complete를 파라미터로 받는다.

마블 다이어그램​

마블 다이어그램은 연산자를 쉽게 이해하기 위해, 옵져버블을 테스트하기위해 도식화된 다이어그램이다. 타이밍과 값의 변화를 한 눈에 파악할 수 있다.

• 위에 줄은 input 이고 연산자를 만나고나면 아래 줄에서 output 이 어떻게 바뀌는지 보여진다.
• 가로 줄은 하나의 옵져버블이다.
• 줄의 | 파이프는 구독 완료를 나타낸다.

소스​

생성​

가장 간단한 옵져버블을 만들고 확인해보자.

const { Observable } = require("rxjs");

const test$ = Observable.create((observer) => {
  console.log("create");
  observer.next(1);
  observer.next(2);
  observer.complete();
  console.log("done");
});

test$.subscribe(
  (item) => {
    console.log(item);
  },
  (error) => {},
  () => {
    console.log("complete");
  },
);

/*
create
1
2
complete
done
*/

간단하지만 이터레이터이면서 구독가능하다는 걸 확인할 수 있다.

구독해제​

옵져버블의 리턴함수로 구독해제 콜백을 지원한다, 콜백이 필요하지 않다면 unsubscribe() 를 호출해주기만 하면 된다.

const test$ = Observable.create((observer) => {
  const interval = setInterval(() => {
    console.log("test");
  }, 1000);

  return () => {
    clearInterval(interval);
  };
});

const subscription = test$.subscribe();
subscription.unsubscribe();

파이퍼블 연산자​

Pipeable 연산자는 옵져버블 인스턴스를 pipe 함수 안에서 다룰 수 있는 연산자이다. 기본적으로 rxjs/operators 라이브러리 안에 들어있다.

const { map } = require("rxjs/operators");

const test$ = Observable.create((observer) => {
  observer.next(1);
  observer.next(2);
  observer.complete();
});

test$.pipe(map((value) => value * 2)).subscribe((item) => console.log(item));

/*
2
4
*/

소스 옵져버블에서 발행된 값을 원하는대로 바꿀 수 있다.

연산자​

of​

args 순서대로 값을 반환한다.

from​

• Observable
• Array
• Promise
• Iterable
• String
• ArrayLike

위 타입을 옵져버블로 변환해준다.

fromEvent​

EventEmitter 클래스의 객체와 조합하거나 브라우저의 이벤트를 옵져버블로 바꿀 때 사용한다.

defer​

팩토리 함수로 옵져버블을 생성한 후 구독한느 시점에 팩토리 함수를 호출해 이미 생성한 옵져버블을 리턴받아 구독한다.

from 과의 차이는 다름과 같다.

• from
  • 프로미스 내부 구현부가 언제 실행되던지 상관 없을 때
  • 이미 실행 중이거나 완료한 프로미스를 옵져버블로 만들 때
• defer
  • 옵져버블을 구독하는 시점에 프로미스를 생성해 구현부가 실행되어야할 때
  • 프로미스 객체 생성 시점이 구독하는 시점이여야할 때

range​

범위 지정 후 그 값을 순서대로 발행한다. 반복문이 필요할 때 사용된다.

interval​

ms 단위로 값을 발행한다.

timer​

파라미터가 하나일 경우 ms 이후에 한 번 값을 발행하고, 두 개일 경우 ms 이후에 두번 째 파라미터만큼 주기적으로 값을 발행한다.

empty​

값 발행 후 중간에 멈춰야하는 상황에 사용한다. 이 함수만 사용하지는 않고 다른 함수나 연산자와 조합해서 complete 함수를 호출해야 할 때 사용된다. 즉, 바로 구독을 완료해야될때 사용된다.

// 상수로 사용된다.
const { EMPTY } = require("rxjs");

never​

아무 것도 하지 않고 옵져버블 생성이 필요할 때 사용된다.

// 상수로 사용된다.
const { NEVER } = require("rxjs");

throwError​

옵져버블로 값을 발행하다가 에러를 발생시키고 종료해야하는 상황에 사용한다.

filter​

주로 파이퍼블 연산자와 연결해서 사용된다.

const { filter } = require("rxjs/operators");
// 1~10 중 짝수 필터
range(1, 10)
  .pipe(filter((x) => x % 2 === 0))
  .subscribe((x) => console.log(x));

first​

처음으로 일치하는 값을 발행한다. 두 번째 인자로 기본 값을 줄 수 있다.

last​

마지막으로 일치하는 값을 발행한다. 두 번째 인자로 기본 값을 줄 수 있다.

take​

정해진 갯수만큼 구독하고 구독을 해제한다. interval 과 같이 무한 반복이 실행되는 연산자와 같이 쓰면 된다.

const { take } = require("rxjs/operators");

interval(1000)
  .pipe(take(3))
  .subscribe((x) => console.log(x));

/*
0
1
2
*/

takeUntil​

특정 이벤트가 발생할 때까지 옵져버블을 구독해야할 때 사용한다. 예시로 보는 게 빠르다.

interval(1000)
  .pipe(
    take(100),
    takeUntil(fromEvent(document.querySelector("#btn"), "click")),
  )
  .subscribe((x) => console.log(x));

takeWhile​

take 와 filter 가 합쳐진 연산자이다.

interval(1000)
  .pipe(takeWhile((x) => x <= 10))
  .subscribe((x) => console.log(x));

takeLast​

Last 의 파라미터 수 만큼 저장해뒀다가 구독 완료시에 일괄적으로 발행한다. 발행하는 값이 [0, 2, 4, 6, 8, 10] 일 때 takeLast(3) 일 경우 내부에 저장된 값은 다음과 같다.

skip​

이름 그대로 n 개만큼의 발행을 건너 뛴다.

skipUntil​

takeUntil 과의 반대로 옵져버블이 실행될 때까지 건너 뛴다.

const time = 1000;
interval(time)
  .pipe(skipUntil(interval(time * 5)), take(2))
  .subscribe((x) => console.log(x));

/*
4
5
*/

skipWhile​

조건을 만족하지 않는 순간부터 값을 발행한다.

debounce​

많이 사용하진 않지만, 디바운스할 옵져버블을 계산값을 리턴해주는 함수를 파라미터로 주면 된다.

debounceTime​

로대쉬나 다른 라이브러리의 debounce 와 같다.

distinct​

중복은 제거하고 발행한다. 값 비교에는 === 연산자가 사용된다.

of({ id: 1 }, { id: 1 }, { id: 2 }, { id: 3 })
  .pipe(
    distinct((data) => data.id),
    map((data) => data.id),
  )
  .subscribe((x) => console.log(x));

/**
1
2
3
*/

두번째 파라미터로 flush 조건을 옵져버블로 넘겨 중복 조건을 초기화시킬 수 있다.

distinctUntilChanged​

중복값이 연속으로 발행된 경우만 제거한다.

of(1, 2, 3, 3, 4, 1)
  .pipe(distinctUntilChanged())
  .subcribe((x) => console.log(x));

/**
1
2
3
4
1
*/

첫번째 파라미터로는 비교함수 (prev, next 를 파라미터로 받는)를 넣어 연속 비교조건을 변경할 수 있다.

두번째 파라미터로는 비교할 값 셀렉터를 변경해줄 수 있다.

of(
  { a: 1, b: 10 },
  { a: 1, b: 10 },
  { a: 2, b: 20 },
  { a: 3, b: 30 },
  { a: 3, b: 30 },
  { a: 2, b: 20 },
)
  .pipe(distinctUntilChanged(null, (data) => data.a))
  .subscribe((x) => console.log(x));

sample​

이건 마블 다이어그램으로 이해하는 게 빠르다.

notifier 옵져버블(x 옵져버블)이 발행되면 이전 최근 값을 발행한다. 값 c 처럼 소스 옵져버블에 새로운 값이 없을 경우 값을 중복으로 발행하지 않는다.

sampleTime​

일정 간격 사이에 있는 최근 값을 발행한다.

pluck​

기본 map 연산자는 뻔하기에 적지 않았다. map 처럼 동작하지만 소스 옵져버블에서 객체를 리턴할 때 객체의 property 를 뽑아낸다.

mergeMap​

switchMap​

concatMap​

scan​

partition​

groupBy​

buffer​

bufferCount​

window​

windowCount​

여담​

• RxJS 의 장점은 이벤트 구독을 취소하고, 모든 이벤트를 하나의 스트림으로 제어할 수 있는 것이라고 생각한다.
• 이해하는데 상당한 기간이 소요되었지만, 웹에선 다음과 같은 상황에서 사용될 것 같다.
  • SPA
  • 에디터와 같이 많은 컴포넌트끼리 통신이 필요할 때
• 이 상황에서도 앵귤러가 아닌 스토어 기반의 리액트, 뷰를 사용한다면 RxJS를 쓸거야 라고 보여주기 위해 사용하는 느낌이 든다.
• 모두가 같이 가기엔 꽤 어려운 개념이다.
• 싱글 스레드 기반의 백엔드에서 더 유용하게 쓰일 수 있을 듯
• learn-rxjs의 예제를 분석하는 게 더 많은 도움이 될 것 같다.