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

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

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

Потоки

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

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

Поток выше изображен в виде марбл-диаграммы, но есть и альтернативный способ визуализации событий — c помощью ASCII кодировки.

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

• UI события, любое взаимодействие пользователя с интерфейсом
• Запросы к серверу
• События устройства, пуш уведомления, системные нотификации
• Веб хуки

Примечание: часто потоки сравнивают с promise. На самом деле, они имеют всего одну общую черту — push стратегию распространения изменений. В остальном это абсолютно разные сущности. Promise не может выдать несколько значений. Он может только выполнить resolve или reject, т.е. иметь только два состояния. Поток же может передавать несколько значений, и может быть повторно использован.

Чем обрабатывать потоки?

Как мы выяснили, наше приложение — инкубатор всевозможных потоков.

1. EventTarget интерфейс обработки различных UI событий в javascript.
2. EventEmitter используется при построении асинхронно-событийной архитектуры в Node.js
3. Функции обратного вызова как вариант обработки однократного асинхронного события, например, чтения файла
4. Promise так же используются для однократных асинхронных вычислений
5. Генераторы — функции с возможностью приостановить свое выполнение на некоторое время, после чего возобновить вновь. Своего рода имплементация корутин в javascript, позволяющая писать асинхронный код в синхронном стиле
6. Web sockets и Web workers

Было бы удобно иметь единый интерфейс взаимодействия с любым средством событийной обработки. И такой интерфейс предоставляет бибилиотека RxJS (Reactive Extensions for Javascript).

RxJS – это Javascript библиотека для трансформации, составления и извлечения асинхронных потоков данных. Она может быть использована как в браузере, так и на стороне сервера.

Поток может испускать три вида событий:

• значение;
• ошибка;
• завершение.

И наша задача их перехватить и описать реакцию на эти три вида асинхронных (или синхронных) событий. При этом мы будем работать с несколькими программными сущностями, первая из которых – наблюдаемость.

Observable

Теперь, когда мы знаем, что такое потоки, давайте поработаем с ними. В RxJS потоки представлены классом Observable. Чтобы создать свой поток, достаточно вызвать конструктор данного класса и передать ему в качестве аргумента функцию подписки:

Через вызов конструктора класса Observable мы создаем новый поток. В качестве аргумента в конструктор мы передали функцию подписки. Функция подписки — это обычная функция, которая в качестве параметра принимает наблюдателя(observer). Сам наблюдатель представляет собой объект, у которого есть три метода:

• next — выбрасывает новое значение в поток
• error — выбрасывает в поток ошибку, после чего поток завершается
• complete — завершает поток

Таким образом, мы создали поток, который испускает два значения и завершается.

При этом существует 2 вида стратегии поведения между производителем и потребителем данных.

При первой (pull-стратегии) место и время реакции на изменение или получение данных определяется потребителем, при второй (push-стратегии) – производителем.

Примером pull-стратегии является обычный вызов функции из того места программы, где требуются те данные которые она возвращает, при этом сама функция ничего не знает о том, кто и откуда ее будет «дергать».

При push-стратегии – наоборот, функция-производитель как бы «обвязывается» обработчиками (потребителями), которые ничего не знают о том, где и когда эти данные будут сгенерированы.
RxJS использует вторую push-стратегию.

Наблюдаемость ленива. Она не начинает испускать значений, пока кто-то не подпишется на нее функцией subscribe в отличие от промисов.

Subscription

Если мы запустим предыдущий код, то ничего не произойдет. Мы лишь создадим новый поток и сохраним ссылку на него в переменную observable, но сам поток так никогда и не испустит ни одного значения. Это происходит потому, что потоки являются “ленивыми” объектами и ничего сами по себе не делают. Для того, чтобы наш поток начал испускать значения и мы могли бы эти значения обрабатывать, нам необходимо начать “слушать” поток. Сделать это можно, вызвав метод subscribe у объекта observable.

Мы определили нашего наблюдателя и описали у него три метода: next, error, complete. Методы просто логируют данные, которые передаются в качестве параметров. Затем мы вызываем метод subscribe и передаем в него нашего наблюдателя. В момент вызова subscribe происходит вызов функции подписки, той самой, которую мы передали в конструктор на этапе объявления нашего потока. Дальше будет выполняться код функции-подписки, которая передает нашему наблюдателю два значения, а затем завершает поток.

Наверняка, у многих возник вопрос, что будет, если мы подпишемся на поток еще раз? Будет все то же самое: поток снова передаст два значения наблюдателю и завершится. При каждом вызове метода subscribe будет происходить обращение к функции-подписке, и весь ее код будет выполняться заново. Отсюда можно сделать вывод: сколько бы раз мы не подписывались на поток, наши наблюдатели получат одни и те же данные.

Пайпы & операторы

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

Например, они могут преобразовать значение и передать его дальше в поток, или могут выполнять роль фильтров и не пропускать какие-либо значения, если они не соответствуют заданному условию.

Примечание: pipe !== subscribe. Метод pipe декларирует поведение потока, но не выполняет подписку. Пока вы не вызовете метод subscribe, ваш поток не начнет работать.

Пример get запрос к серверу и вывод информации в список с помощью rx js

const getPosts$ = fromEvent(requestButton, 'click')
   .pipe(switchMap(el => ajax('https://jsonplaceholder.typicode.com/posts').pipe(
       map(v => v.response), switchMap((v) => from(v).pipe(
           map(arr => arr), map(arr => arr.title)
       ))
   )))
getPosts$.subscribe((value) => createList(value))
const  createList = (value) => {
   let li = document.createElement("li")
   li.innerText = value
   containerDiv.appendChild(li)
}

1. fromEvent — принимает DOM элемент и событие
2. switchMap — Проецирует каждое исходное значение в Observable, который объединяется с выходным Observable, выдавая значения только из самого последнего проецированного Observable.
3. ajax — get запрос
4. from — принимает массив, разбивает его на элементы

Рисование на canvas с помощью rx js

const moveCanvas$ = fromEvent(canvas, 'mousemove')
const clearCanvas$ = fromEvent(clearButton, 'click')
const drawInCanvas = (event) => {
   ctx.fillStyle = 'green'
   ctx.fillRect(event.offsetX, event.offsetY, 3, 3)
}
moveCanvas$.subscribe((e) => drawInCanvas(e));
clearCanvas$.subscribe(() => ctx.clearRect(0,0,canvas.width, canvas.height))

1. canvas — dom element
2. clearButton — кнопка очистки canvas

C помощью rx js создаём 2 потока на событие click и mousemove, и подписываемся на них, функция drawcanvas рисует линию, а метом clearRect очищает canvas.