[RxJS] Subjects

A Subject is like an Observable, but can multicast to many Observers. Subjects are like EventEmitters: they maintain a registry of many listeners.

subject는 대표적인 Hot Observable이다. 또한 모든 Subject는 옵저버(Observer)이며, 구독(subscribe)하면 등록된 모든 옵저버에게 멀티 캐스팅된다. RxJS에서 제공하는 공식 예제를 봐도 잘 나와있다.

import { Subject } from 'rxjs';

const subject = new Subject<number>();

subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});

subject.next(1);
subject.next(2);

// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2

보시다시피 옵저버A와 옵저버B에게 멀티캐스팅 되서 값을 값이 전달(공유)되는 것을 볼 수 있다.
이게 Subject의 핵심이다. 이제 RxJS에서 지원하는 멀티캐스팅 연산자를 살펴보겠다.

---

[cf] Hot Observable?

옵저버블이 푸시하는 값을 여러 옵저버에 멀티캐스팅하는 옵저버블이다. 이게 무슨 말인가 하겠지만 간단하게 설명하자면, 간단하게 설명하자면 구독(Subscribe)하는 대상들과 같이 데이터를 함께 공유한다.

[cf] Cold Observable?

핫 옵저버블과는 달리 멀티캐스팅을 지원하지 않는 옵저버블이다. 독립적으로 동작하며 푸시하는 값이 여러 옵저버에 공유되지 않는다.

---

BehaviorSubject

BehaviorSubject(value);

위 코드를 보면 신기한 점이 있다. 데이터가 전달되기 전에 구독(subscribe)되어 있다면 구독(subscribe)된 대상에게 초깃값을 전달하고 있다. 이게 BehaviorSubject이다.

여기서 옵저버A에만 'start'가 전달된 이유는 옵저버A에게 데이터가 전달되기 전에 이미 구독하고 있었기 때문이다. 데이터 전달 후 옵저버B는 구독을 하였기 때문에 'start'라는 초깃값이 전달되지 않는다.
위에서 방금까지 초깃값이라고 표현했지만 정확하게 따지면 옵저버에게 방출하는 최신값이 맞는 표현입니다.

처음부터 최신값이라고 하면 이해가 잘 안되기 때문에... 여튼 옵저버B 입장에서 보면, 옵저버B가 구독을 하였을 때의 최신값은 3이다.. 그래서 구독을 하자마자 초깃값(최신값)을 받는 모습을 볼 수 있다.

그리고 재밌는 기능이 하나 더 있다.

데이터가 전달된 후부터는 구독(subject)된 시점과 관계 없이 마지막 상태를 전달한다. 여기서 옵저버B의 마지막 상태는 complete된 상태이기 때문에 'complete'된 이후지만 observerB Complete가 전달된 것이다.

ReplySubject

ReplaySubject(bufferSize?: number, windowTime?: number);

BehaviorSubject와 매우 유사하지만(새 구독자(옵저버)에게 이전 값을 보낼 수 있다는 점), 옵저버블의 실행의 일부를 기록할 수 있다. 다시 말하자면 ReplySubject는 bufferSize만큼 데이터를 저장하고 있다가 구독(subscribe)될 때 가장 최근 bufferSize만큼 데이터를 전달한다. 또한 windowTime을 전달받아 내부적으로 저장할 데이터 개수와 저장할 시간을 지정할 수 있다.

예제를 보면 이해가 더 빨리된다. 간단하게 말해서 첫번째 옵셔널 값에 3을 넣었고(당연한 말이겠지만 제네릭도 지원한다. 아까 위 예제에서도 써도 된다) 최근 3개의 값이 저장된다. 그리고 새로 구독하는 옵저버B에게 전달해 주는 것이다. 사실 const subject = new ReplaySubject<number>(); 이렇게 쓰는거랑 동일하다. 파라미터가 없이 값을 전달하면 전달된 모든 데이터를 공유 받는다. 이걸 이용해 보면 재밌는 상황도 만들 수 있다.

옵저버A는 complete를 만나 정상적으로 종료되었지만, 데이터가 전달된 후부터는 구독(subscribe) 시점과 상관없이 마지막 상태를 전달하기 때문에 버퍼에 담긴 3개의 값을 전달하고 끝이난다.

windowTime을 사용하는 예제는 RxJS 홈페이지를 참고 부탁드린다. (굳이..)

---

AsyncSubject

AsyncSubject()

complete가 호출되었을 때 마지막 하나의 값만 next하고 종료된다.

complete가 되지 않았다면 어떤 데이터도 전달하지 않는다.