Кастомные операторы 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". Если вам интересно узнать о курсе подробнее, приглашаем на день открытых дверей онлайн, где преподаватель расскажет о формате обучения и программе.