10 необходимых для понимания функций RxJS с примерами

Эта статья была рецензирована Флорианом Рапплом и Морицем Крёгером . Спасибо всем рецензентам SitePoint за то, что сделали контент SitePoint как можно лучше!

По мере роста интереса к функционально-реактивному программированию (FRP) RxJS стала одной из самых популярных библиотек JavaScript для программирования в этой парадигме. В этой статье мы рассмотрим то, что я считаю десятью функциями, которые должны знать RxJS.

Примечание. Эта статья предполагает знакомство с основами RxJS, как представлено в статье « Введение в функционально-реактивное программирование с использованием RxJS» .

Реактивное программирование

Реактивное программирование — это парадигма программирования, которая рассматривает потоки данных, называемые Observables, в качестве основных единиц программирования.

Потоки — или Observables , на языке RxJS — аналогичны прослушивателям событий: оба ждут, что что-то произойдет, и уведомляют вас, когда это происходит. Серия асинхронных уведомлений, которые вы получаете от слушателя onClick, является прекрасным примером потока данных.

Иными словами, Observable — это не что иное, как массив, который заполняется с течением времени .

Элементы этого массива могут поступать практически из любого места: файловая система, события DOM, вызовы API или даже преобразованные синхронные данные, например массивы. По сути, реактивное программирование — это не что иное, как использование Observables в качестве строительных блоков ваших программ.

Отношение к массивам

Массивы просты, потому что их содержимое является окончательным, если явно не изменено. В этом смысле нет ничего по существу временного в массиве.

Наблюдаемое, с другой стороны, определяется временем. Максимум, что вы можете знать о потоке, это то, что он получил [1, 2, 3] до сих пор. Вы не можете быть уверены, что когда-либо получите 4 или не получите — и это решает источник данных, а не ваша программа.

Взаимосвязь между потоками и массивами настолько глубока, что большинство реактивных расширений происходят из мира функционального программирования, где операции со списками — это хлеб с маслом.

Разогрев до RxJS

Рассмотрим слишком общее приложение для дел. Давайте посмотрим, как выглядит проблема отображения только имен незавершенных задач пользователя с помощью RxJS:

 const task_stream = // Makes a stream of all the tasks in the database getTasks(). // Get tasks only for this user filter((task) => task.user_id == user_id). // Get tasks that are incompleted filter((task) => !task.completed). // Only get name of task map((task) => task.name) /* Tasks look like this: task = { user_id : number, completed : boolean, name : string } */ 

Пока это не что иное, как массивы , но демонстрирует функциональный вкус реактивного программирования.

Его декларативный характер становится понятным с добавлением более сложных, «реальных» функций. Допустим, мы хотим:

• Отклонить запрос в ответ на выбор пользователя для просмотра завершенных или незавершенных задач;
• Отправляйте запрос на последний выбор только каждую секунду, чтобы не тратить пропускную способность, если пользователь быстро меняет свой выбор;
• Повторите неудачные запросы до трех раз; и
• Перерисовывать представление только тогда, когда сервер отправляет другой ответ с прошлого раза.

 const task_stream = parameter_stream. debounce(1000). map((parameter) => { getTasks(). retry(3). filter((task) => task.user_id === user_id). filter((task) => task.completed === parameter). map((task) => task.name) }). flatMap(Rx.Observable.from). distinctUntilChanged(). update() 

Шаг за шагом:

 source. map((user) => user.website) 

 source. map((user) => user.website). filter((website) => (website.endsWith('net') || website.endsWith('org')); }) 

 source. map((user) => user.website). filter((website) => (website.endsWith('net') || website.endsWith('org'))). reduce((data, website) => { return { count : data.count += 1, name_length : data.name_length += website.length } }, { count : 0, name_length : 0 }) 

 source. map((user) => user.website). filter((website) => (website.endsWith('net') || website.endsWith('org'))). scan((data, website) => { return { count : data.count += 1, name_length : data.name_length += website.length } }, { count : 0, name_length : 0 }). take(2); 

 source. map((user) => user.website). filter((website) => (website.endsWith('net') || website.endsWith('org'))). scan((data, website) => { return { count : data.count += 1, name_length : data.name_length += website.length } }, { count : 0, name_length : 0 }). takeWhile((data) => data.count < 3) 

 const source = // Take a Promise and convert it to an Observable Rx.Observable.fromPromise(makeRequest(ENDPOINT)) // Flatten Promise .flatMap(Rx.Observable.from); 

 Rx.Observable.from([1, 2, 3]). subscribe( onNext (value) => console.log(`Next: ${value}`)) // Prints: // Next: 1 // Next: 2 // Next: 3 

 const source = // Create an Observable emitting the VALUE or REJECTION of a Promise... Rx.Observable.fromPromise(makeRequest(ENDPOINT)) // ...And turn it into a new Observable that emits every item of the // array the Promise resolves to. .flatMap(Rx.Observable.from) 

 source1. concat(source2). subscribe( onNext(value) => console.log(`Next: ${value}`)) // Prints 'Source 1' values first, THEN 'Source 2' source1. merge(source2). subscribe( onNext(value) => console.log(`Next: ${value}`)) // INTERLEAVES 'Source 1' and 'Source 2' values 

 // User's selection for either POSTS or USERS data const endpoint_stream = Rx.Observable.fromEvent(select_endpoint, 'click'). map(event => event.target). map(target => (target.options[target.selectedIndex].text.toLowerCase())); // Which item ID the user wants to retrieve const id_stream = Rx.Observable.fromEvent(select_id, 'click'). map(event => event.target). map(target => (target.options[target.selectedIndex].text)); // Emits a pair of the most recent selections from BOTH streams // when EITHER emits a value const complete_endpoint_stream = endpoint_stream.combineLatest(id_stream); 

 // stream1 : Emits 1 // stream2 : Emits 1 combined : Emits [1, 1] // stream2: Emits 2 combined : Emits [1, 2] // stream2: Emits 3 combined : Emits [1, 3] 

 // stream1 : Emits A // stream2 : Emits 1 zipped : Emits [A, 1] // stream2: Emits 2 zipped : Emits NOTHING // stream2: Emits 3 zipped : Emits NOTHING // stream1: Emits B zipped : Emits [B, 2] // stream1: Emits C zipped : Emits [C, 3] 

 source1. takeUntil(source2);