Какие операторы для работы с потоками вы знаете?

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

Операторы создания (Creation Operators): Используются для создания нового потока данных.

• of(): Создает поток, который испускает фиксированный набор значений.

  // Из RxJS
  import { of } from 'rxjs';
  
  const source = of(1, 2, 3); // Поток, испускающий 1, 2, 3
  

• from(): Превращает различные источники данных (Promise, Iterable, ArrayLike) в поток.

  // Из RxJS
  import { from } from 'rxjs';
  
  const source = from([10, 20, 30]); // Поток из массива
  

• interval(): Создает поток, который испускает последовательные числа через заданный интервал времени.

  // Из RxJS
  import { interval } from 'rxjs';
  
  const source = interval(1000); // Поток, испускающий числа каждую секунду
  

• timer(): Создает поток, который испускает одно значение после заданного времени, опционально с последующими значениями через интервал.
• fromEvent(): Создает поток из DOM-событий или EventEmitter.

Операторы трансформации (Transformation Operators): Изменяют значения, испускаемые потоком.

• map(): Преобразует каждое значение, испускаемое потоком, применяя к нему заданную функцию.

  // Из RxJS
  import { of } from 'rxjs';
  import { map } from 'rxjs/operators';
  
  const source = of(1, 2, 3);
  const result = source.pipe(map(x => x * 2)); // Поток с значениями 2, 4, 6
  

• pluck(): Извлекает свойство из каждого объекта, испускаемого потоком.
• scan(): Применяет аккумуляторную функцию к каждому значению потока и испускает промежуточные результаты.
• mergeMap() / flatMap(): Проецирует каждое значение исходного потока в новый поток (Inner Observable) и объединяет (сглаживает) их в один выходной поток. Позволяет выполнять асинхронные операции для каждого значения.

  // Из RxJS
  import { of } from 'rxjs';
  import { mergeMap } from 'rxjs/operators';
  
  const source = of(1, 2);
  const result = source.pipe(mergeMap(val => of(val + 'a', val + 'b'))); // Поток с значениями '1a', '1b', '2a', '2b'
  

• switchMap(): Похож на mergeMap, но при испускании нового значения исходным потоком отменяет работу предыдущего внутреннего потока. Полезен для автодополнения или запросов с отменой.
• concatMap(): Похож на mergeMap, но гарантирует, что внутренние потоки будут выполнены последовательно в порядке их создания.

Операторы фильтрации (Filtering Operators): Выборочно пропускают значения, испускаемые потоком.

• filter(): Пропускает значения из потока только если они удовлетворяют заданному условию.

  // Из RxJS
  import { of } from 'rxjs';
  import { filter } from 'rxjs/operators';
  
  const source = of(1, 2, 3, 4);
  const result = source.pipe(filter(x => x % 2 === 0)); // Поток с значениями 2, 4
  

• take(): Пропускает только первые N значений из потока.
• takeUntil(): Пропускает значения из потока пока другой поток не испустит значение (сигнал останова).
• skip(): Пропускает первые N значений из потока.
• debounceTime(): Испускает значение только после того, как поток прекратил испускать значения в течение заданного времени. Полезен для предотвращения слишком частых действий (например, ввода в поле поиска).
• distinctUntilChanged(): Испускает значение только если оно отличается от предыдущего испущенного значения.

Операторы комбинирования (Combination Operators): Объединяют несколько потоков в один.

• combineLatest(): Объединяет значения из нескольких потоков, испуская массив последних значений из каждого потока при каждом изменении одного из входящих потоков.
• zip(): Объединяет значения из нескольких потоков, испуская массив значений по индексу (первое из первого, первое из второго и т.д.).
• merge(): Объединяет несколько потоков, испуская значения из них по мере их появления.
• concat(): Объединяет несколько потоков, выполняя их последовательно (сначала первый до завершения, затем второй и т.д.).

Операторы обработки ошибок (Error Handling Operators): Определяют поведение при ошибках в потоке.

• catchError(): Перехватывает ошибку в потоке и позволяет продолжить работу потока, либо заменить его другим потоком.

  // Из RxJS
  import { throwError } from 'rxjs';
  import { catchError, of } from 'rxjs/operators';
  
  const source = throwError(() => new Error('Что-то пошло не так'));
  const result = source.pipe(catchError(err => of('Ошибка перехвачена'))); // Поток с значением 'Ошибка перехвачена'
  

• retry(): Пытается повторно подписаться на поток N раз при возникновении ошибки.

Операторы завершения (Completion Operators): Определяют поведение при завершении потока.

• finalize(): Выполняет заданное действие при завершении (успешном или с ошибкой) или отмене подписки на поток.
• tap(): Позволяет выполнять побочные эффекты (например, логирование) без изменения значений потока.

Это далеко не полный, но достаточно важный список операторов, с которыми часто приходится работать во Frontend-разработке, особенно при использовании реактивного подхода.