Операторы и метод pipe()

Для предварительного преобразования отправляемых объектом Observable данных или преобразования и управления самими Observable используются специальные функции — операторы.

from([7, 21, 10])
    .pipe(map((num) => (num <= 10 ? 1 : 0)))
    .subscribe((vl) => console.log(vl));

Здесь в map() каждое число равное или меньше 10 заменяется на 1, если больше — на 0.

Все RxJS операторы подразделяются на категории. Так, различают операторы:

• Создания (of, from, fromEvent, interval);
• Преобразования (map, scan, buffer);
• Фильтрации (filter, take, skip, distinct);
• Обработки ошибок (catchError, retry, onErrorResumeNext);
• Условия (skipUntil, skipWhile, takeUntil, takeWhile);
• Математические (min, max, count);
• Утилиты (tap, delay);
• Для Connectable Observable (share, shareReplay, publish).

pipe()

Начиная с версии 5.5 в библиотеке RxJS большинство операторов, которые применяются к Observable, объединяются в методе pipe(). Метод вызывается у объекта, а операторы передаются ему в качестве аргументов через запятую в порядке их вызова.

from([30, 41, 60])
    .pipe(
        filter((num) => num % 10 === 0),
        map((num) => num * 2)
    )
    .subscribe((vl) => console.log(vl));

Методу pipe() передаются только те операторы, которые принимают Observable и возвращают Observable. Они находятся в rxjs/operators и называются pipeable operators (англ.).

В RxJS также есть одноименная утилита pipe(), используемая для создания переиспользуемых композиций операторов.

const doubleFilter = <T, R>(
    fn: (value: T, index: number) => R
) => pipe(filter(fn), filter(fn));

Пример приведен только для наглядности. Практическое использование такого оператора не эффективно.

Комментарии