RxJS로 옵저버블을 만드는 방법

 

RxJS로 옵저버블을 만드는 다양한 방법을 배워봅니다.

원문 링크입니다 : https://www.thisdot.co/blog/creating-observables-in-rxjs

Creating Observables in RxJS - This Dot Labs

Observable은 RxJS의 기초입니다.
RxJS와 관련된 모든 것은 Observable을 중심으로 이루어집니다.
이 아티클에서는 RxJS에서 제공하는 Observable을 생성하는 다양한 방법을 살펴보겠습니다.

 

RxJS에서 Observable을 생성하는 두 가지 주요 방법이 있습니다.
Subject 및 Operator 입니다.
우리는 이 두 가지를 모두 살펴볼 것입니다!

Observable이 뭔가요?

먼저 Observable이 뭔지 알아봅시다.

Observable은 동기 또는 비동기로 여러 값을 Observer에 푸시하는 인수가 없는 함수와 같습니다.

다소 혼란스러운 정의이므로 4개의 값을 Observer에 푸시하는 Observable의 매우 기본적인 예를 들어 보겠습니다.

const obs$ = Observable.create((observer) => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  setTimeout(() => observer.next(4), 1000);
});

console.log("before subscribe");
const observer = obs$.subscribe((v) => console.log("received: ", v));
console.log("after subscribe");

위의 예에서

우리는 Observable을 생성하고

Observable을 구독하는 즉시 Observer에게 1, 2, 3을 보내도록 지시합니다.

이것이 동기 호출입니다.

그러나 4는 구독 후 로그인한 후 발생하는 1초 후까지 전송되지 않으므로 이를 비동기 작업으로 만듭니다.

출력에서 이를 확인할 수 있습니다.

before subscribe
received:  1
received:  2
received:  3
after subscribe 
received:  4

Observer가 값 푸시를 완료했음을 알릴 때까지 Observer는 값을 계속 수신합니다.
위의 예를 수정하면 실제로 작동하는 것을 볼 수 있습니다.

const obs$ = Observable.create((observer) => {
  observer.next(1);
  observer.next(2);
  observer.complete();
  observer.next(3);
  setTimeout(() => observer.next(4), 1000);
});

console.log("before subscribe");
obs$.subscribe((v) => console.log("received: ", v));
console.log("after subscribe");

우리는 observer.complete()에 대한 호출을 추가했습니다.
Observer.next(2) 이후에 Observer가 값 푸시를 완료했음을 Observer에 알립니다.

결과를 봅시다.

before subscribe 
received:  1
received:  2
after subscribe

Observer에 값 3과 4를 푸시하려고 해도 Observer가 수신하지 않는 것을 볼 수 있습니다.

지금까지 정적 create 생석 메서드를 사용하여 Observable을 생성하는 방법을 설명했습니다.

 

이제 Subject와 Operator를 사용하여 Observable을 만드는 방법을 살펴보겠습니다.

Creating Observables with Subjects

Subject는 EventEmitter와 Observable의 조합으로 생각할 수 있습니다.

그들은 둘 다처럼 행동합니다.

 

Observer는 Subject를 구독하여 Subject가 푸시하는 값을 받을 수 있으며,

Subject는 새 값을 각 Observer에 직접 푸시하거나

각 Observer에게 Subject가 값 푸시를 완료했음을 알릴 수 있습니다.

 

RxJS가 우리에게 제공하는 Subject에는 4가지 유형이 있습니다.

차례로 살펴보겠습니다.

Subject

Subject는 Observable을 생성하는 데 사용할 수 있는 가장 기본적인 Subject의 타입입니다.

사용이 매우 간단하며 구독하는 모든 Subscriber에게 값을 푸시하는 데 사용할 수 있습니다.

각 Observer는 Observer가 구독한 후 Subject가 푸시한 값만 수신합니다.

 

const subject$ = new Subject();

const observerA = subject$.subscribe((v) => console.log("Observer A: ", v));
const observerB = subject$.subscribe((v) => console.log("Observer B: ", v));

subject$.next(1);

const observerC = subject$.subscribe((v) => console.log("Observer C: ", v))

subject$.next(2);

먼저 Subject를 생성한 다음 Subject(Observable)에서 수신한 각 값을 기록할 두 개의 Observer를 생성합니다.

1. Subject에게 값 1을 푸시하도록 지시합니다.
2. 그런 다음 Subject에서 수신한 각 값을 기록하는 ObserverC를 만듭니다.
3. 마지막으로 Subject에게 값 2를 푸시하도록 지시합니다.

이제 다음 출력을 살펴봅시다.

Observer A:  1
Observer B:  1
Observer A:  2
Observer B:  2
Observer C:  2

ObserverA와 ObserverB는 둘 다 1을 받았지만 ObserverC는 2만 받은 것을 볼 수 있습니다.
기본적인 Subject의 Observer는 구독한 후에 푸시된 값만 수신한다는 점을 강조합니다!

BehaviorSubject

우리가 사용할 수 있는 또 다른 타입의 Subject는 BehaviorSubject입니다.

한 가지 주요 차이점을 제외하고는 기본 Subject와 정확히 동일하게 동작합니다.

현재 값을 알고 있습니다.

Subject가 새 값을 푸시하면 이 값을 내부적으로 저장합니다.

새로운 Observer가 BehaviorSubject를 구독하면

BehaviorSubject는 Observer에게 푸시한 마지막 값을 즉시 보냅니다.

 

Subject에 사용한 예제를 사용하여 BehaviorSubject를 사용하도록 변경하면

이 함수가 실제로 작동하는 것을 볼 수 있습니다.

const behaviorSubject$ = new BehaviorSubject();

const observerA = behaviorSubject$.subscribe((v) => console.log("Observer A: ", v));
const observerB = behaviorSubject$.subscribe((v) => console.log("Observer B: ", v));

behaviorSubject$.next(1);

const observerC = behaviorSubject$.subscribe((v) => console.log("Observer C: ", v))

behaviorSubject$.next(2);

 

차이점을 확인하기 위해 출력을 살펴보겠습니다.

Observer A:  1
Observer B:  1
Observer C:  1
Observer A:  2
Observer B:  2
Observer C:  2​

ObserverC가 1이 푸시된 후 BehaviorSubject를 구독했음에도 불구하고 값 1이 전송되었음을 알 수 있습니다.

ReplaySubject

ReplaySubject는 푸시한 값을 기억하고 구독한 새 Observer에게 즉시 보낼 수 있다는 점에서 BehaviorSubject와 매우 유사합니다.

그러나 기억해야 하는 값의 수를 지정할 수 있으며,

이 모든 값을 구독하는 각각의 새 Observer에게 보냅니다.

 

위의 예를 약간 수정하여 이 기능이 동작하는 것을 보겠습니다.

const replaySubject$ = new ReplaySubject(2); // 2 - number of values to store

const observerA = replaySubject$.subscribe((v) => console.log("Observer A: ", v));

replaySubject$.next(1);
replaySubject$.next(2);
replaySubject$.next(3);

const observerB = replaySubject$.subscribe((v) => console.log("Observer B: ", v))

replaySubject$.next(4);

이번에는 ReplaySubject가 4개의 값을 Observer에 푸시하도록 할 것입니다.
우리는 또한 그것이 방출한 두 개의 최신 값을 항상 저장해야 한다고 말합니다.

출력을 살펴보겠습니다.

Observer A:  1
Observer A:  2
Observer A:  3
Observer B:  2
Observer B:  3
Observer A:  4
Observer B:  4

우리는 ObserverA가 처음 3개의 값을 완벽하게 잘 수신한다는 것을 알 수 있습니다.

그런 다음 ObserverB는 ReplaySubject를 구독하고,ReplaySubject는 이전에 마지막으로 푸시한 두 값 2와 3 값을 즉시 보냅니다.

그런 다음 두 Observer는 다음 값인 4를 올바르게 받습니다.

AsyncSubject

AsyncSubject는 Subject와 동일한 메서드를 모두 노출하지만 다르게 동작합니다.

Observer에게 푸시하도록 지시받은 마지막 값만을 전송하며, Subject가 완료되었을 때만 이 작업을 수행합니다(complete() 호출).

따라서 Observer는 Subject가 완료될 때만 값을 수신하고, 이후에 구독하는 Observer는 완료될 때 푸시한 값을 즉시 수신합니다.

const asyncSubject$ = new AsyncSubject(2);

const observerA = asyncSubject$.subscribe((v) =>
  console.log("Observer A: ", v)
);

asyncSubject$.next(1);
asyncSubject$.next(2);

const observerB = asyncSubject$.subscribe((v) =>
  console.log("Observer B: ", v)
);

asyncSubject$.next(3);
asyncSubject$.complete();

const observerC = asyncSubject$.subscribe((v) =>
  console.log("Observer C: ", v)
);

 

출력은 다음과 같습니다.

Observer A:  3
Observer B:  3
Observer C:  3

ObserverA는 값이 푸시되기 전에 구독했지만, 3개의 값중 마지막 값인 3만 수신했음을 알 수 있습니다.
ObserverC도 AsyncSubject가 완료(complete)된 후 구독했음에도, 즉시 값 3을 수신했음을 알 수 있습니다.

Operators로 Observable 만들기

Observable을 생성하는 다른 방법은 RxJS의 연산자를 활용하는 것입니다.

이러한 연산자는 의도에 따라 분류할 수 있습니다.

이 기사에서는 Observable을 생성하는 Creation Operators를 살펴보겠습니다.

여기에서 이러한 연산자 목록을 볼 수 있습니다.
http://reactivex.io/rxjs/manual/overview.html#creation-operators

ajax

ajax는 AJAX 요청을 처리하기 위한 Observable을 생성하는 Operator입니다.

URL, 헤더 등이 포함된 요청 객체 또는 URL 문자열을 취합니다.

요청이 완료되면 Observable이 완료됩니다.

이를 통해 AJAX 요청을 만들고 반응적으로 처리할 수 있습니다.

const obs$ = ajax("https://api.github.com/users?per_page=2");
obs$.subscribe((v) => console.log("received: ", v.response));

 

출력은 다음과 같습니다.

received:  (2) [Object, Object]

bindCallback

bindCallback을 사용하면 일반적으로 콜백 접근 방식을 사용하는 모든 함수를 Observable로 변환할 수 있습니다.
설명만으론 이해하기 상당히 어려울 수 있으므로 예를 들어 설명하겠습니다.

// Let's say we have a function that takes two numbers, multiplies them
// and passes the result to a callback function we manually provide to it
function multiplyNumbersThenCallback(x, y, callback) {
  callback(x * y);
}

// We would normally use this function as shown below
multiplyNumbersThenCallback(3, 4, (value) =>
  console.log("Value given to callback: ", value)
);

// However, with bindCallback, we can turn this function into
// a new function that takes the same arguments as the original
// function, but without the callback function
const multiplyNumbers = bindCallback(multiplyNumbersThenCallback);

// We call this function with the numbers we want to multiply
// and it returns to us an Observable that will only push 
// the result of the multiplication when we subscribe to it
multiplyNumbers(3, 4).subscribe((value) =>
  console.log("Value pushed by Observable: ", value)
);

bindCallback을 사용하면 Callback API를 사용하는 함수를 가져와
구독할 수 있는 Observable을 생성하는 반응적인 함수로 변환할 수 있습니다.

defer

defer를 사용하면 Observer가 구독할 때만 Observable을 생성할 수 있습니다.

각 Observer에 대해 새로운 Observable을 생성합니다.

즉, 동일한 Observable을 공유하는 것처럼 보이더라도 공유하지 않습니다.

const defferedObs$ = defer(() => of([1, 2, 3]));

const observerA = defferedObs$.subscribe((v) => console.log("Observer A: ", v));
const observerB = defferedObs$.subscribe((v) => console.log("Observer B: ", v));​

출력은 다음과 같습니다.

Observer A:  (3) [1, 2, 3]
Observer B:  (3) [1, 2, 3]

두 Observer는 동일한 값이 푸시된 Observable을 받았습니다.

이들은 동일한 값을 푸시하더라도 실제로는 다른 Observable입니다.

다음 예제를 수정하여 defer가 각 Observer에 대해 서로 다른 Observable을 생성한다는 것을 설명할 수 있습니다.

let numOfObservers = 0;
const defferedObs$ = defer(() => {
  if(numOfObservers === 0) {
    numOfObservers++;
    return of([1, 2, 3]);
  }

  return of([4,5,6])
});

const observerA = defferedObs$.subscribe((v) => console.log("Observer A: ", v));
const observerB = defferedObs$.subscribe((v) => console.log("Observer B: ", v));​

첫 번째 Observer에게 [1, 2, 3]의 Observable을 제공하고 다른
모든 Observers [4, 5, 6]에 제공하도록 defer 객체를 변경했습니다.

이제 출력에서 다른 점을 볼 수 있습니다.

Observer A:  (3) [1, 2, 3]
Observer B:  (3) [4, 5, 6]

empty

empty 연산자는 값을 푸시하지 않고, 구독할 때 즉시 완료되는 Observable을 생성합니다.

const obs$ = empty();
obs$.subscribe((v) => console.log("received: ", v));

값을 푸시하지 않으므로 출력이 없습니다.

from

from은 강력한 연산자입니다.
거의 모든 것을 Observable로 변환하고

이러한 원천(source) 자체를 기반으로 지능적인 방식으로 이러한 소스의 값을 푸시합니다.

배열과 제너레이터의 iterable 두 가지 예를 들어보겠습니다.

const obs$ = from([1,2,3]);
obs$.subscribe((v) => console.log("received: ", v));

배열의 경우 from은 배열의 각 요소를 가져와 별도로 푸시합니다.

received:  1
received:  2
received:  3

마찬가지로 제너레이터에서 iterable을 사용하여 각 값을 개별적으로 얻습니다.

function* countToTen() {
  let i = 0;
  while(i < 11) {
    yield i;
    i++;
  }
}

const obs$ = from(countToTen());
obs$.subscribe((v) => console.log("received: ", v));

fromEvent

fromEvent Operator는
웹 페이지의 모든 클릭과 같이 지정된 이벤트 타겟에서 발생한 지정된 유형의 모든 이벤트를 푸시하는 Observable을 생성합니다.

우리는 이것을 매우 쉽게 설정할 수 있습니다:

const obs$ = fromEvent(document, "click");
obs$.subscribe(() => console.log("received click!"));

페이지를 클릭할 때마다 "received click!"이 기록됩니다.

received click!
received click!

fromEventPattern

fromEventPattern은 발생한 이벤트에 대해 동작한다는 점에서 fromEvent 연산자와 유사합니다.

그러나 두 가지 인수가 필요합니다.

addHandler 함수 인수 및 removeHandler 함수 인수 입니다.

addHandler 함수는 Observable이 구독될 때 호출되며
구독한 Observer는 addHandler 함수에 설정된 모든 이벤트를 수신합니다.

removeHandler 함수는 Observer가 Observable에서 구독 취소할 때 호출됩니다.

실제 코드를 보는 것보다 설명이 복잡할 수 있습니다.
페이지에서 발생하는 모든 클릭을 가져오려는 위의 예를 사용하겠습니다.

function addHandler(handler) {
  document.addEventListener('click', handler)
}

function removeHandler(handler) {
  document.removeEventListener('click', handler)
}

const obs$ = fromEventPattern(addHandler, removeHandler);
obs$.subscribe(() => console.log("received click!"));

페이지를 클릭할 때마다 "received click!"이 기록됩니다.

received click!
received click!

generate

generater를 사용하면 멈출 때를 알려주는 조건(predicate)과 함께
전달한 인수를 기반으로 푸시할 값을 생성하는 Observable을 만들 수 있습니다.

const obs$ = generate(
  1,
  (x) => x < 11,
  (x) => x++
)

obs$.subscribe((v) => console.log("received: ", v));

이것은 다음을 출력합니다.

received:  0
received:  1
received:  2
received:  3
received:  4
received:  5
received:  6
received:  7
received:  8
received:  9
received:  10

interval

interval 연산자는 설정된 시간 간격으로 새 값을 푸시하는 Observable을 만듭니다.
아래 예제는 매초 새 값을 푸시하는 Observable을 만드는 방법을 보여줍니다.

const obs$ = interval(1000);
obs$.subscribe((v) => console.log("received: ", v));

매초마다 새 값을 기록합니다.

received:  0
received:  1
received:  2

never

never 연산자는 새 값을 푸시하지 않고 오류가 발생하지 않으며 완료되지 않는 Observable을 만듭니다.
다른 Observable과 함께 테스트하거나 구성하는 데 유용할 수 있습니다.

const obs$ = never();
// This never logs anything as it never receives a value
obs$.subscribe((v) => console.log("received: ", v));

of

of 연산자는

제공한 인수와 동일한 순서로 값을 푸시한 다음 완료하는 Observable을 생성합니다.
from 연산자와 달리 배열의 모든 요소를 ​​가져와서 각각 푸시하지는 않습니다.

대신 전체 배열을 하나의 값으로 푸시합니다.

const obs$ = of(1000, [1,2,4]);
obs$.subscribe((v) => console.log("received: ", v));

출력은 다음과 같습니다.

received:  1000
received:  (3) [1, 2, 4]

range

range 연산자는 지정된 두 값 사이의 값을 순서대로 푸시하는 Observable을 만듭니다.
10까지 세는 함수를 range 연산자를 사용하여 생성할 수 있는 방법을 보겠습니다.

const obs$ = range(0, 10);
obs$.subscribe((v) => console.log("received: ", v));

출력은 다음과 같습니다.

received:  0
received:  1
received:  2
received:  3
received:  4
received:  5
received:  6
received:  7
received:  8
received:  9
received:  10

throwError

throwError 연산자는 값을 푸시하지 않고 즉시 오류 알림을 푸시하는 Observable을 생성합니다.
Observer가 Observable을 구독할 때 Observable이 던진 오류를 정상적으로 처리할 수 있습니다.

const obs$ = throwError(new Error("I've fallen over"));
obs$.subscribe(
  (v) => console.log("received: ", v),
  (e) => console.error(e)
);

출력은 다음과 같습니다.

Error: I've fallen over

timer

timer는 지정된 지연이 끝날 때까지 값을 푸시하지 않는 Observable을 만듭니다.
또한 초기 지연 후 각 간격에서 증가하는 값을 푸시하는 간격 시간을 알릴 수 있습니다.

const obs$ = timer(3000, 1000);
obs$.subscribe((v) => console.log("received: ", v));

출력은 3초 후에 발생하기 시작하고 각 로그는 1초 간격입니다.

received:  0
received:  1
received:  2
received:  3

---

해당 글이 RxJS로 작업할 때 도움이 되었으면 합니다.

bindCallback 및 fromEvent와 같이 미묘한 사용 사례에 매우 유용할 수 있는 몇 가지 생성 연산자가 있습니다.