Кастомные операторы RxJS

Библиотека RxJS благодаря широкому выбору операторов по праву считается крайне мощным инструментом в арсенале разработчика. В этом посте я хочу представить вам концепцию кастомных операторов RxJS с примерами реализации.

Библиотека RxJS благодаря широкому выбору операторов по праву считается крайне мощным инструментом в арсенале разработчика. Недавно я подготовил несколько самописных операторов, чтобы повысить удобство повторного использования некоторых комбинаций операторов. В этом посте я хочу представить вам концепцию кастомных операторов RxJS с примерами реализации.

Оператор идентификации

Оператор RxJS — это всего лишь функция, которая берет некие наблюдаемые (observable) данные в качестве входных и возвращает результирующий поток. Следовательно, задача написания кастомного оператора RxJS сводится к написанию обычной функции JavaScript (TypeScript). Начнем с базового оператора идентификации (identity), который просто зеркалирует наблюдаемые исходные данные:

import { interval, Observable } from "rxjs"; import { take } from "rxjs/operators";  const source$ = interval(1000).pipe(take(3));  function identity<T>(source$: Observable<T>): Observable<T> {   return source$; }  const results$ = source$.pipe(identity);  results$.subscribe(console.log);    // console output: 0, 1, 2

Далее напишем кастомный оператор с кое-какой элементарной логикой.

Оператор логирования

Следующий кастомный оператор выполняет побочное действие (логирует значения в консоли) для каждого значения исходного потока:

<>Copy import { interval, Observable } from "rxjs"; import { take, tap } from "rxjs/operators";  const source$ = interval(1000).pipe(take(3));  function log<T>(source$: Observable<T>): Observable<T> {   return source$.pipe(tap(v => console.log(`log: ${v}`))); }  const results$ = source$.pipe(log);  results$.subscribe(console.log);    // console output: log: 0, log: 1, log: 2

В основе результирующего потока лежат данные source$, которые видоизменяются посредством применения встроенных операторов в составе метода pipe.

Фабрика оператора

В некоторых сценариях для кастомного оператора полезно указать контекст. Для этого можно определить функцию, возвращающую оператор. Аргументы фабрики входят в лексическую область видимости оператора:

import { interval, Observable } from "rxjs"; import { take, tap } from "rxjs/operators";  const source$ = interval(1000).pipe(take(3));  function logWithTag<T>(tag: string): (source$: Observable<T>) => Observable<T> {   return source$ =>     source$.pipe(tap(v => console.log(`logWithTag(${tag}): ${v}`))); }  const results$ = source$.pipe(logWithTag("RxJS"));  results$.subscribe(console.log);    // console output: logWithTag(RxJS): 0, logWithTag(RxJS): 1, logWithTag(RxJS): 2

Описание возвращаемого типа можно упростить, воспользовавшись функцией MonoTypeOperatorFunction библиотеки RxJS. Кроме того, с помощью статической функции pipe можно сократить определение оператора:

import { interval, MonoTypeOperatorFunction, pipe } from "rxjs"; import { take, tap } from "rxjs/operators";  const source$ = interval(1000).pipe(take(3));  function logWithTag<T>(tag: string): MonoTypeOperatorFunction<T> {   return pipe(tap(v => console.log(`logWithTag(${tag}): ${v}`))); }  const results$ = source$.pipe(logWithTag("RxJS"));  results$.subscribe(console.log);    // console output: logWithTag(RxJS): 0, logWithTag(RxJS): 1, logWithTag(RxJS): 2

Другие полезные советы по RxJS можно почитать здесь.

Уникальная для наблюдателя лексическая область видимости

Функция фабрики оператора вызывается лишь один раз в момент определения потока. В результате у всех наблюдателей будет общая лексическая область видимости:

import { interval, MonoTypeOperatorFunction, pipe } from "rxjs"; import { take, tap } from "rxjs/operators";  const source$ = interval(1000).pipe(take(3));  function tapOnce<T>(job: Function): MonoTypeOperatorFunction<T> {   let isFirst = true;    return pipe(     tap(v => {       if (!isFirst) {         return;       }        job(v);       isFirst = false;     })   ); }  const results$ = source$.pipe(tapOnce(() => console.log("First value emitted")));  results$.subscribe(console.log); results$.subscribe(console.log);    // console output: First value emitted, 0, 0, 1, 1, 2, 2

Чтобы у каждого наблюдателя была уникальная лексическая область видимости, можно применить функцию defer:

import { defer, interval, MonoTypeOperatorFunction } from "rxjs"; import { take, tap } from "rxjs/operators";  const source$ = interval(1000).pipe(take(3));  function tapOnceUnique<T>(job: Function): MonoTypeOperatorFunction<T> {   return source$ =>     defer(() => {       let isFirst = true;        return source$.pipe(         tap(v => {           if (!isFirst) {             return;           }            job(v);           isFirst = false;         })       );     }); }  const results$ = source$.pipe(tapOnceUnique(() => console.log("First value emitted")));  results$.subscribe(console.log); results$.subscribe(console.log);    // console output: First value emitted, 0, First value emitted, 0, 1, 1, 2, 2

Другой способ решения задачи tapOnce рассматривается в одном из моих предыдущих постов.

Практические примеры

Оператор firstTruthy:

import { MonoTypeOperatorFunction, of, pipe } from "rxjs"; import { first } from "rxjs/operators";  const source1$ = of(0, "", "foo", 69);  function firstTruthy<T>(): MonoTypeOperatorFunction<T> {   return pipe(first(v => Boolean(v))); }  const result1$ = source1$.pipe(firstTruthy());  result1$.subscribe(console.log);  // console output: foo

Оператор evenMultiplied:

import { interval, MonoTypeOperatorFunction, pipe } from "rxjs"; import { filter, map, take } from "rxjs/operators";  const source2$ = interval(10).pipe(take(3));  function evenMultiplied(multiplier: number): MonoTypeOperatorFunction<number> {   return pipe(     filter(v => v % 2 === 0),     map(v => v * multiplier)   ); }  const result2$ = source2$.pipe(evenMultiplied(3));  result2$.subscribe(console.log);    // console output: 0, 6

Оператор liveSearch:

import { ObservableInput, of, OperatorFunction, pipe  } from "rxjs"; import { debounceTime, delay, distinctUntilChanged, switchMap } from "rxjs/operators";  const source3$ = of("politics", "sport");  type DataProducer<T> = (q: string) => ObservableInput<T>;  function liveSearch<R>(   time: number,   dataProducer: DataProducer<R> ): OperatorFunction<string, R> {   return pipe(     debounceTime(time),     distinctUntilChanged(),     switchMap(dataProducer)   ); }  const newsProducer = (q: string) =>   of(`Data fetched for ${q}`).pipe(delay(2000));  const result3$ = source3$.pipe(liveSearch(500, newsProducer));  result3$.subscribe(console.log);    // console output: Data fetched for sport

Заключение

Типовые комбинации операторов RxJS можно вынести в кастомные операторы и многократно использовать их в будущем при реализации аналогичного функционала. Применение дженериков обеспечивает корректное приведение типов выходных значений, обрабатываемых в дальнейшей pipe-последовательности.

Живой пример: [смотрите в оригинале]

Надеюсь, вам понравился мой пост и вы узнали что-то новое.

Перевод материала подготовлен в рамках курса «JavaScript Developer. Professional«. Если вам интересно узнать о курсе подробнее, приглашаем на день открытых дверей онлайн, где преподаватель расскажет о формате обучения и программе.