Зачем использовать materialize и dematerialize операторы, и что такое Notification в RxJS?19.04.2022 11:46

Немного визуализации:

Работа оператора в коде:

// 1 пример
of("one”) // эмитится одно значение, и завершается поток
.pipe(
	materialize(), // конвертируем в Notification oбъект
  dematerialize()) // конвертируем Notification в исходное значение потока
.subscribe(x => console.log(x));

Output: "one" // получили то, что изначально эмитил поток

// 2 пример
throwError(() => new Error("error”)) // эмитим ошибку
.pipe(
	materialize(), // конвертируем в Notification oбъект
  dematerialize()) // конвертируем Notification в исходное значение потока
.subscribe(x => console.log(x));

Output: // ничего не логируется next обработчиком, 
ERROR Error: error // видим обычную ошибку

А нужно ли оно мне?

Мы вкратце рассмотрели, что делают операторы materialize и dematerialize. Но где же они могут пригодиться?

Первое, где их можно использовать — во время дебага. Например, вам нужно добавить задержку во время эмита ошибки.

Если это сделать следующим образом:  

throwError(() => new Error("message”))
.pipe(delay(3000)) // добавляем задержку
.subscribe({
	next: (x) => { console.log(`${x}`); },
	error: (err) => { console.log(`${err}`); }
	complete: () => { console.log('complete'); }
})

То в этом случае ошибку вы увидите моментально, задержка не сработает. Это происходит потому, что оператор delay работает только для next выбросов.

Но если использовать изученные нами операторы, то ошибка выводится после заданного времени задержки.

throwError(() => new Error('message'))
      .pipe(
      	materialize(), // ошибка конвертируется в Notification объект, это позволяет нам применить delay    y
      	delay(5000), // добавляем задержку
        dematerialize()) // конвертируем обратно в ошибку
      .subscribe({
        next: (x) => { console.log(`${x}`); },
        error: (err) => { console.log(`${err}`); },
        complete: () => { console.log('complete'); },
      });
      
Output: Error: message // сработал error обработчик через 5 секунд

Все из-за того, что ошибка конвертируется в Notification объект с помощью materialize, это позволяет нам применить оператор delay.

После (с помощью dematerialize оператора) мы получаем первоначальную ошибку.

Исходный код

Где же еще они пригодятся?

Скажем, нужно объединить 2 observable. Как только один из них завершится, то результирующий observable тоже должен завершиться.

Если реализовать это с помощью оператора merge:

const result$ = merge(this.first$, this.second$)
      .subscribe({
        next: (x) => { console.log(`${x}`); },
        complete: () => { console.log('complete'); },
      });
 
    this.first$.next(1);
 
    this.second$.complete(); // завершаем один из потоков
 
    this.first$.next(2); // результирующий поток все еще не завершен

Output:
1
2

Мы не увидим в output «complete». Это случилось из-за того, что merge просто так работает:»The output Observable only completes once all input Observables have completed». И один из способов справиться с этим — использовать операторы materialize и dematerialize:

 const result$ = merge(
      this.first$.pipe(materialize()), // конвертируем в Notification 1 поток 
      this.second$.pipe(materialize()) // конвертируем в Notification 2 поток 
      ) 
      .pipe(dematerialize()) // возвращаемся к исходному значению
      .subscribe({
        next: (x) => { console.log(`${x}`); },
        complete: () => { console.log('complete'); },
      });
 
    this.first$.next(1); // Notification {kind: "N"}
 
    this.second$.complete(); // завершаем один из потоков - Notification {kind: "C"}
 
    this.first$.next(2); // этот next уже на сработает

Output: 
1 // next от first$
complete // complete от second$ — завершился результирующий поток

Это то, чего мы добивались. Догадываетесь, почему сработало?

Мы знаем, что при использовании merge оператора результирующий поток не завершится при завершении только одного из переданных в него observable. А materialize оператор конвертирует любое значение observable в Notification объект, что воспринимается как «next» выброс.

Это позволило нам получить «complete» событие от this.second$ observable в pipe (…).
И уже с помощью dematerialize оператора мы получаем исходное значение observable — завершение потока.

Исходный код

Заключение

Я постаралась привести несколько примеров использования materialize/dematerialize, чтобы понять, на что они способны и где полезны.

Уверена, что эти операторы помогут вам лучше понимать и использовать все возможности RxJS, и вы найдете им интересные способы применения.

А вы раньше использовали их?

Полезные ссылки

https://rxjs.dev/api/operators/materialize
https://rxjs.dev/api/operators/dematerialize
https://rxjs.dev/api/index/class/Notification
https://docs.w3cub.com/rxjs/api/index/class/notification
https://docs.w3cub.com/rxjs/api/operators/materialize
https://docs.w3cub.com/rxjs/api/operators/dematerialize
https://docs.w3cub.com/rxjs/api/index/function/merge

© Habrahabr.ru