Реактивное программирование в TypeScript: библиотека RxJS

Реактивное программирование в TypeScript: полное руководство по RxJS

Реактивное программирование — это парадигма разработки, ориентированная на асинхронные потоки данных и распространение изменений. В экосистеме TypeScript наиболее мощным инструментом для реализации этой парадигмы является библиотека RxJS (Reactive Extensions for JavaScript). RxJS позволяет работать с событиями, асинхронными запросами и состояниями приложения как с последовательностями данных, которые можно трансформировать, комбинировать и фильтровать с помощью декларативных операторов.

Основные концепции RxJS

В основе RxJS лежат три ключевые сущности:

• Observable — источник данных, который может генерировать значения с течением времени. Аналог массива, но элементы появляются асинхронно.
• Observer — потребитель данных, который реагирует на три типа событий: next (новое значение), error (ошибка) и complete (завершение потока).
• Subscription — объект, представляющий выполнение Observable. Позволяет отменять подписку и освобождать ресурсы.

import { Observable } from 'rxjs';

const observable = new Observable<number>(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => {
    subscriber.next(4);
    subscriber.complete();
  }, 1000);
});

const subscription = observable.subscribe({
  next: (value) => console.log(`Получено: ${value}`),
  error: (err) => console.error(`Ошибка: ${err}`),
  complete: () => console.log('Поток завершен')
});

// Отмена подписки через 500 мс
setTimeout(() => subscription.unsubscribe(), 500);

Операторы RxJS: основа реактивных трансформаций

Операторы — это чистые функции, которые принимают один Observable и возвращают новый Observable, не изменяя исходный. Они делятся на несколько категорий:

• Создающие: of, from, interval, fromEvent
• Трансформирующие: map, pluck, scan, buffer
• Фильтрующие: filter, take, skip, distinctUntilChanged
• Комбинирующие: merge, concat, combineLatest, zip
• Обработка ошибок: catchError, retry, retryWhen

import { fromEvent, interval } from 'rxjs';
import { map, filter, debounceTime, distinctUntilChanged } from 'rxjs/operators';

// Пример: реактивный поиск с debounce
const searchInput = document.getElementById('search') as HTMLInputElement;
const input$ = fromEvent(searchInput, 'input').pipe(
  map(event => (event.target as HTMLInputElement).value),
  filter(text => text.length >= 3),
  debounceTime(300),
  distinctUntilChanged()
);

input$.subscribe(searchTerm => {
  console.log(`Поиск: ${searchTerm}`);
  // Выполнить HTTP-запрос
});

Subject: горячие и холодные потоки

Subject — это особый тип Observable, который одновременно является и наблюдателем, и наблюдаемым. Он позволяет мультикастировать (распространять) одно и то же значение множеству подписчиков. Основные разновидности:

• Subject — базовый, передаёт только новые значения
• BehaviorSubject — хранит последнее значение и выдаёт его новым подписчикам
• ReplaySubject — буферизирует N последних значений
• AsyncSubject — выдаёт только последнее значение после завершения потока

import { BehaviorSubject } from 'rxjs';

// Управление состоянием приложения
interface AppState {
  user: string | null;
  isAuthenticated: boolean;
}

const initialState: AppState = {
  user: null,
  isAuthenticated: false
};

const state$ = new BehaviorSubject<AppState>(initialState);

// Компонент подписывается на изменения
state$.subscribe(state => {
  console.log('Состояние обновлено:', state);
});

// Где-то в коде происходит обновление
state$.next({
  user: 'admin',
  isAuthenticated: true
});

Практический пример: реактивный HTTP-клиент

Рассмотрим типичный сценарий — загрузка данных с сервера с возможностью отмены и повторной попытки:

import { from, Observable, throwError } from 'rxjs';
import { map, catchError, retry, timeout } from 'rxjs/operators';

interface User {