RxJS: Subject, Behavior Subject и Replay Subject

Subject’ы очень полезны при множественных подписках или в случаях, когда источник потока сложно трансформировать в observable переменную. Но ими легко можно злоупотребить. Поэтому мы рассмотрим основные типы Subject’ов и в каких случаях их стоит использовать.

Subject — это некий гибрид в RxJS, который выступает одновременно в роли observable и observer. Это означает, что данные можно пушить в subject, а все кто подписаны на него получат данные.

Помимо простого subject, есть еще несколько специальных типов: async subjects, behavior subjects и replay subjects.

Как работает subject

Использовать «сабджект» достаточно просто, необходимо лишь создать объект класса.

const mySubject = new Rx.Subject();

На него можно подписаться в нескольких местах и «сабджект» будет хранить список подписок на него.

const mySub = mySubject.subscribe(x => console.log(`${x} ${x}`));
const mySub2 = mySubject.subscribe(x => console.log(x.toUpperCase()));

Данные передаются с помощью метода next.

mySubject.next('Hello golosay.net!');

// Hello golosay.net! Hello golosay.net!
// HELLO GOLOSAY.NET!

Когда данные пушатся в «сабджект», он пробежится по всему списку подписок и передаст полученные данные. Достаточно просто, не правда ли?

Тогда вот еще пример, который продемонстрирует вышеописанный функционал.

const mySubject = new Rx.Subject();

mySubject.next(1);

const subscription1 = mySubject.subscribe(x => {
  console.log('Подписка 1:', x);
});

mySubject.next(2);

const subscription2 = mySubject.subscribe(x => {
  console.log('Подписка 2:', x);
});

mySubject.next(3);

subscription1.unsubscribe();

mySubject.next(4);

В результате мы получим:

Подписка 1: 2
Подписка 1: 3
Подписка 2: 3
Подписка 2: 4

Обратите внимание, что пуш с цифрой 1 утерян, так как транслировался раньше, чем была объявлена подписка на «сабджект». Для решения подобных проблем используются Behavior Subjects и Replay Subjects. О них я расскажу далее.

Ошибки и их обработка

Если у “сабджекта” вызвать метод error(), все подписчики смогут обработать вызванную ошибку:

const mySubject = new Rx.Subject();

const sub1 = mySubject.subscribe(null, err =>
  console.log('От sub1: ', err.message)
);

const sub2 = mySubject.subscribe(null, err =>
  console.log('От sub2: ', err.message)
);

mySubject.error(new Error('О нет! Что-то пошло не так...'));

// От sub1: О нет! Что-то пошло не так...
// От sub2: О нет! Что-то пошло не так...

Multicasting

Настоящая мощь subject’а раскрывается при мультикастинге, так называют технику, когда «сабджект» передается в observable переменную как observer. И когда observable транслирует событие, данные передаются всем подписчикам subject’а.

Вот пример, в котором observable trickleWords, транслирует слово из массива каждых 750мс подписчикам subject’а.

const mySubject = new Rx.Subject();
const words = ['Hot Dog', 'Pizza', 'Hamburger'];

const trickleWords = Rx.Observable.zip(
  Rx.Observable.from(words),
  Rx.Observable.interval(750),
  word => word
);

const subscription1 = mySubject.subscribe(x => {
  console.log(x.toUpperCase());
});

const subscription2 = mySubject.subscribe(x => {
  console.log(
    x
      .toLowerCase()
      .split('')
      .reverse()
      .join('')
  );
});

trickleWords.subscribe(mySubject);

Вот что выведется в консоль:

HOT DOG
god toh
PIZZA
azzip
HAMBURGER
regrubmah

Subject в Observable

Метод asObservable используется для преобразования subject’а в простой observable. Это может быть полезным в тех случаях, когда необходимо получать данные подписчикам, но необходимо запретить добавление данных в «сабджект».

const mySubject = new Rx.Subject();
const myObs = mySubject.asObservable();

mySubject.next('Hello');
myObs.next('World!'); // TypeError: myObs.next is not a function

Replay Subjects

Теперь вернемся к ситуации, когда мы теряли данные, если они транслировались раньше объявления подписки. Replay subject создан для решения таких проблем. Он может хранить последние траслирующееся данные в буфере и отдавать их при новой подписке. Количество хранимых трансляций можно указать в атрибуте функции. Вот пример работы с буфером на два последних эмита:

const mySubject = new Rx.ReplaySubject(2);

mySubject.next(1);
mySubject.next(2);
mySubject.next(3);
mySubject.next(4);

mySubject.subscribe(x => {
  console.log('От первого sub:', x);
});

mySubject.next(5);

mySubject.subscribe(x => {
  console.log('От второго sub:', x);
});

Вот что увидим в консоли:

От первого sub: 3
От первого sub: 4
От первого sub: 5
От второго sub: 4
От второго sub: 5

Как видите, последних два эмита данных не теряются до подписки, а после подписки передаются в штатном режиме.

Behavior Subjects

Это такие же «сабджекты» как и replay, только в них нельзя выставить размер буфера. Он по умолчанию всегда сохраняет данные последнего эмита.

const mySubject = new Rx.BehaviorSubject('Куку!');

mySubject.subscribe(x => {
  console.log('От первого sub:', x);
});

mySubject.next(5);

mySubject.subscribe(x => {
  console.log('от второго sub:', x);
});

И вот какой мы увидим результат:

От первого sub: Куку!
От первого sub: 5
От второго sub: 5

Вот мы и рассмотрели самые простые примеры использования «сабджектов» для понимания принципа их действий и отличий между ними.

RxJS: все о Subjects, Behavior Subjects и Replay Subjects