Библиотека RxJS

Реактивное программирование - это парадигма асинхронного программирования, связанная с потоком данных и распространением изменений.

RxJS (Reactive Extensions for JavaScript) - это библиотека для реактивного программирования с использованием наблюдаемых, которая облегчает составление асинхронного или основанного на обратном вызове кода (RxJS Docs).

RxJS предоставляет реализацию типа Observable, которая необходима до тех пор, пока тип не станет частью языка и пока его не поддержат браузеры. Библиотека также предоставляет служебные функции для создания и работы с наблюдаемыми. Эти служебные функции могут быть использованы для:

• Преобразование существующего кода для асинхронных операций в наблюдаемые
• Перебирая значения в потоке
• Отображение значений в разные типы
• Фильтрация потоков
• Составление нескольких потоков

Наблюдаемые функции создания

RxJS предлагает ряд функций, которые можно использовать для создания новых наблюдаемых. Эти функции могут упростить процесс создания наблюдаемых из таких вещей, как события, таймеры, обещания и так далее. Например:

Создание наблюдаемого из обещания

Код TypeScript:

import { from } from 'rxjs';
// Create an Observable out of a promise
const data = from(fetch('/api/endpoint'));
// Subscribe to begin listening for async result
data.subscribe({
 next(response) { console.log(response); },
 error(err) { console.error('Error: ' + err); },
 complete() { console.log('Completed'); }
});

Демонстрация в реальном времени:

См. Pen observable_from_promise от w3resource ( @ w3resource ) в CodePen .

Создайте наблюдаемую из счетчика

Код TypeScript:

import { interval } from 'rxjs';
// Create an Observable that will publish a value on an interval
const secondsCounter = interval(1000);
// Subscribe to begin publishing values
secondsCounter.subscribe(n =>
  console.log(`It's been ${n} seconds since subscribing!`));

Демонстрация в реальном времени:

См. Pen observable_from_counter от w3resource ( @ w3resource ) в CodePen .

Создание наблюдаемого из события

Код TypeScript:

import { fromEvent } from 'rxjs';
const el = document.getElementById('my-element');
// Create an Observable that will publish mouse movements
const mouseMoves = fromEvent(el, 'mousemove');
// Subscribe to start listening for mouse-move events
const subscription = mouseMoves.subscribe((evt: MouseEvent) => {
  // Log coords of mouse movements
  console.log(`Coords: ${evt.clientX} X ${evt.clientY}`);
  // When the mouse is over the upper-left of the screen,
  // unsubscribe to stop listening for mouse movements
  if (evt.clientX < 40 && evt.clientY < 40) {
    subscription.unsubscribe();
  }
});

Демонстрация в реальном времени:

См. Pen observable_from_event от w3resource ( @ w3resource ) на CodePen .

Наблюдаемый по запросу Ajax

Код TypeScript:

import { ajax } from 'rxjs/ajax';
// Create an Observable that will create an AJAX request
const apiData = ajax('/api/data');
// Subscribe to create the request
apiData.subscribe(res => console.log(res.status, res.response));

Демонстрация в реальном времени:

См. Pen observable_from_AjaxRequest от w3resource ( @ w3resource ) в CodePen .

операторы

Операторы - это функции, которые строятся на основе наблюдаемых, чтобы обеспечить сложные манипуляции с коллекциями. Например, RxJS определяет операторы, такие как map (), filter (), concat () и flatMap ().

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

Оператор карты

Код TypeScript:

import { map } from 'rxjs/operators';
const nums = of(1, 2, 3);
const squareValues = map((val: number) => val * val);
const squaredNums = squareValues(nums);
squaredNums.subscribe(x => console.log(x));
// Logs
// 1
// 4
// 9

Демонстрация в реальном времени:

См. Pen map_Operator от w3resource ( @ w3resource ) в CodePen .

Вы можете использовать каналы, чтобы связать операторов вместе. Трубы позволяют объединить несколько функций в одну функцию. Функция pipe () принимает в качестве аргументов функции, которые вы хотите объединить, и возвращает новую функцию, которая при выполнении запускает составные функции в последовательности.

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

В качестве примера рассмотрим функцию pipe ниже:

Код TypeScript:

import { filter, map } from 'rxjs/operators';
const nums = of(1, 2, 3, 4, 5);
// Create a function that accepts an Observable.
const squareOddVals = pipe(
  filter((n: number) => n % 2 !== 0),
  map(n => n * n)
);
// Create an Observable that will run the filter and map functions
const squareOdd = squareOddVals(nums);
// Subscribe to run the combined functions
squareOdd.subscribe(x => console.log(x));

Демонстрация в реальном времени:

См. Описание функции Pen Observable_pipe от w3resource ( @ w3resource ) в CodePen .

Этот код выше можно сделать короче, потому что функция канала также является методом в RxJS Observable, это показано ниже:

Код TypeScript:

import { filter, map } from 'rxjs/operators';
const squareOdd = of(1, 2, 3, 4, 5)
  .pipe(
    filter(n => n % 2 !== 0),
    map(n => n * n)
  );
// Subscribe to get values
squareOdd.subscribe(x => console.log(x));

Демонстрация в реальном времени:

См. Функцию Pen mapped_pipe от w3resource ( @ w3resource ) в CodePen .

Обработка ошибок

В дополнение к обработчику error (), который вы предоставляете при подписке, RxJS предоставляет оператор catchError, который позволяет обрабатывать известные ошибки в наблюдаемом рецепте.

Например, предположим, что у вас есть наблюдаемая, которая делает запрос API и сопоставляется с ответом от сервера. Если сервер возвращает ошибку или значение не существует, выдается ошибка. Если вы поймаете эту ошибку и предоставите значение по умолчанию, ваш поток продолжит обрабатывать значения, а не выдавать ошибку.

Пример использования оператора catchError для этого приведен ниже:

Код TypeScript:

import { ajax } from 'rxjs/ajax';
import { map, catchError } from 'rxjs/operators';
// Return "response" from the API. If an error happens,
// return an empty array.
const apiData = ajax('/api/data').pipe(
  map(res => {
    if (!res.response) {
      throw new Error('Value expected!');
    }
    return res.response;
  }),
  catchError(err => of([]))
);
apiData.subscribe({
  next(x) { console.log('data: ', x); },
  error(err) { console.log('errors already caught... will not run'); }
});

Демонстрация в реальном времени:

См. Pen catchError_Operator от w3resource ( @ w3resource ) в CodePen .

Повторить не удалось заметным

Если оператор catchError предоставляет простой путь восстановления, оператор повтора позволяет повторить неудачный запрос.

Используйте оператор retry перед оператором catchError. Он повторно подписывается на исходный наблюдаемый источник, который затем может повторно выполнить полную последовательность действий, приведших к ошибке. Если это включает в себя HTTP-запрос, он будет повторять этот HTTP-запрос.

Следующий код преобразует предыдущий пример, чтобы повторить запрос, прежде чем перехватить ошибку:

Код TypeScript:

import { ajax } from 'rxjs/ajax';
import { map, retry, catchError } from 'rxjs/operators';
const apiData = ajax('/api/data').pipe(
  retry(3), // Retry up to 3 times before failing
  map(res => {
    if (!res.response) {
      throw new Error('Value expected!');
    }
    return res.response;
  }),
  catchError(err => of([]))
);
apiData.subscribe({
  next(x) { console.log('data: ', x); },
  error(err) { console.log('errors already caught... will not run'); }
});

Демонстрация в реальном времени:

См. Pen retry_operator от w3resource ( @ w3resource ) на CodePen .

Предыдущая: Наблюдаемые
Далее: Наблюдаемые в Angular