Sharereplay rxjs что это
Noveo
RxJS для продолжающих
Библиотека RxJS — мощный инструмент для реализации обработки синхронного и асинхронного кода. Мы предполагаем, что читатель, приступающий к прочтению этого материала, уже освоил базовые концепты — это наблюдатель (Observer), наблюдаемое (Observable), операторы (Operators) и подписка (Subscription), а также уже использовал RxJS в проекте и хочет расширить свой опыт и экспертизу по этому вопросу.
Когда начинаешь работать с RxJS, часто встречаешь термин «потоки». Прочитав пару-тройку статей, мы, вдохновившись идеологией реактивного программирования (все есть поток), начинаем строить «потоковую» архитектуру своего приложения. А используя термин «потоки», мы быстро попадаем в когнитивную ловушку: нам кажется что Observable работает (или должен работать) как EventEmitter…
…но на самом деле порядок вещей следующий: по умолчанию Observable работает как функция — то есть вызывается каждый раз при подписке (unicast), но может работать и как EventEmitter — регистрировать количество подписчиков и вещать всем (multicast).
Вам, как бармену, надоело работать попугаем, и вы решили автоматизировать процесс. Допустим, вы купили телевизор и стали с каким-то интервалом выводить на нем бегущую строку с сообщением, как дойти до туалета, а сами спрятались под стойку. Теперь каждый приходящий посетитель вынужден садиться за стол и ожидать очередного сообщения по ТВ. Немного неудобно, зато все присутствующие видят сообщение разом! Широковещательное ТВ: типичный горячий Observable.
Unicast (Cold) and Multicast (Hot) Observables
Наблюдаемые (Observables) по способу выполнения делятся на две категории: “холодные” (unicast) и “горячие” (multicast).
Как мы уже знаем, «холодный» создает собственный поток для каждого нового подписчика и выдает значения лично для него.
В то время как «горячий» регистрирует слушателей и вещает всем.
Библиотека RxJS даёт возможность преобразовать холодный поток в горячий. Но прежде чем говорить о преобразованиях, необходимо рассказать о специальных объектах, которые позволяют это сделать.
4 типа Subjects
В RxJS существуют объекты, которые одновременно являются и наблюдателем, и наблюдаемым — Subjects. Subject’ы придумали для того, чтобы передавать значения сразу нескольким подписчикам. Поскольку Subject’ы по умолчанию являются «горячими», то мы можем быть уверены, что у нас создается один поток, значения из которого будут получать наши подписчики.
Subject
Пример: В примере ниже мы создаем Subject и кладем в него значения от 0 до 4 через каждую секунду. Создаем подписку A сразу после объявления Subject, а подписку B через 2,5 секунды.
BehaviorSubject
— всегда возвращает последнее значение из потока, при этом обязательно нужно указывать начальное значение. Также имеет статический метод getValue, который возвращает статическое значение без подписок и потоков.
Пример использования: создание стейта для приложения.
Пример: В этом примере мы создаем точно такой же поток, как и с простым Subject, но только теперь используем BehaviorSubject. Как мы можем видеть, при старте подписки A новых значений еще не было получено, поэтому возвращается начальное значение 42. Когда создается подписка B, сначала возвращается последнее значение, и после продолжается с новыми значениями.
ReplaySubject
— имеет возможность указания размера буфера.
Пример использования: кеширование запросов к API.
Пример: в данном случае мы указали размеры буфера 2. При создании подписки B мы сначала получили значения 0 и 1.
AsyncSubject
— возвращает последнее значение после завершения потока. Частой ошибкой и непониманием работы с этим типом Subject’a является отсутствие вызова функции завершения потока.
Пример: в этом примере мы кладем значения в AsyncSubject от 0 до 4 каждую секунду. Делаем подписку A сразу и подписку B через 2,5 секунды. И дополнительно вызываем метод complete через 4 секунды, тем самым преждевременно завершая наш поток. В результате в консоли получим значения 3 для подписки A и B.
From Cold to Hot Observable
Все наши преобразования холодного в горячее мы будем проводить на одном примере, который указан ниже.
В данном примере мы создаем холодный поток, который каждую секунду выдаёт значение, начиная с 0 и увеличивая каждый раз на 1.
Multicast
Оператор, который позволяет преобразовать холодное наблюдаемое в горячее, — это оператор multicast. Этот оператор в качестве параметра принимает объект типа subject. Это может быть любой из 4-х видов subject’ов в зависимости от ваших целей.
Обратите внимание, что наш Observable возвращает объект с типом ConnectableObservable. Этот объект похож на обычный Observable, за исключением одного момента: он начинает производить элементы не тогда, когда на него подписываются, а только тогда, когда на нем вызвана функция .connect()
Поэтому представленный выше пример ничего не делает. Для того, чтобы наш Observer начал выдавать значение, нам необходимо вызвать метод connect.
Publish
Следующий оператор, который мы рассмотрим, — это оператор publish. Является сокращенным типом записи оператора multicast(new Subject()). Также помимо оператора publish существуют другие операторы, который упрощают запись и сводят ее к одной строке.
refCount
Каждый раз вызывать метод connect получается достаточно проблематично, мы можем с легкостью забыть его вызвать. Для того, чтобы об этом не волноваться, существует оператор refCount. Этот оператор сам следит за количеством подписчиков и вызывает метод connect(), когда количество подписчиков меняется с 0 на 1, и делает отписку, когда количество подписчиков меняется с 1 на 0.
Все вышеперечисленные методы являются устаревшими. Им на смену пришли новые операторы, которые еще больше упрощают процесс преобразования Observable (текущая версия на момент написания статьи RxJS 6.5.3).
Share
Первый оператор — это Share. Это вызов операторов publish и refCount, но только вызов будет состоять из одного оператора.
Также существует оператор shareReplay. Это вызов операторов publishReplay и refCount. С ним есть особенность использования в зависимости от версии библиотеки. Подробнее о его непростой судьбе можно прочитать здесь: https://blog.angularindepth.com/rxjs-whats-changed-with-sharereplay-65c098843e95.
Memory Leaks
Вспомни про историю с баром. К сожалению, каждый посетитель занимает в баре место, а каждый включенный телевизор кушает электричество. Следовательно, когда бару пора закрываться, эти ресурсы нужно уметь освобождать.
При использовании RxJS в ваших компонентах очищать ресурсы (делать .unsubscribe()) надо при каждом дестрое компонента, в котором они занимались, иначе они останутся в памяти, и с браузером вашего пользователя будет тоже самое, что и с баром!
Про необходимость и способы контроля за подписками в Ангуляре вы можете прочитать в этой замечательной статье: https://medium.com/ngx/why-do-you-need-unsubscribe-ee0c62b5d21f.
Conclusions
Вот мы и стали на шаг ближе к званию мастера RxJS! Осталось лишь освоить Schedulers, написание своих кастомных операторов и их тестирование. Всем удачи на этом извилистом пути!
Заходит однажды тестировщик в бар.
Забегает в бар.
Пролезает в бар.
Танцуя, проникает в бар.
Прокрадывается в бар.
Врывается в бар.
Прыгает в бар
и заказывает:
кружку пива,
2 кружки пива,
0 кружек пива,
999999999 кружек пива,
ящерицу в стакане,
–1 кружку пива,
qwertyuip кружек пива.
Первый реальный клиент заходит в бар и спрашивает, где туалет. Бар вспыхивает пламенем, все погибают.
Если вы нашли ошибку, пожалуйста, выделите фрагмент текста и нажмите Ctrl+Enter.
Повтор неудачных HTTP-запросов в Angular
Организация доступа к серверным данным — это основа почти любого одностраничного приложения. Весь динамический контент в таких приложениях загружается с бэкенда.
В большинстве случаев HTTP-запросы к серверу работают надёжно и возвращают желаемый результат. Однако в некоторых ситуациях запросы могут оказываться неудачными.
Представьте себе, как кто-то работает с вашим веб-сайтом через точку доступа в поезде, который несётся по стране со скоростью 200 километров в час. Сетевое соединение при таком раскладе может быть медленным, но запросы к серверу, несмотря на это, делают своё дело.
А что если поезд попадёт в туннель? Тут высока вероятность того, что связь с интернетом прервётся и веб-приложение не сможет «достучаться» до сервера. В этом случае пользователю придётся перезагрузить страницу приложения после того, как поезд выедет из туннеля и соединение с интернетом восстановится.
Перезагрузка страницы способна оказать воздействие на текущее состояние приложения. Это значит, что пользователь может, например, потерять данные, которые он ввёл в форму.
Вместо того чтобы просто смириться с тем фактом, что некий запрос оказался неудачным, лучше будет несколько раз его повторить и показать пользователю соответствующее уведомление. При таком подходе, когда пользователь поймёт, что приложение пытается справиться с проблемой, он, скорее всего, не станет перезагружать страницу.
Материал, перевод которого мы сегодня публикуем, посвящён разбору нескольких способов повторения неудачных запросов в Angular-приложениях.
Повтор неудачных запросов
Давайте воспроизведём ситуацию, с которой может столкнуться пользователь, работающий в интернете из поезда. Мы создадим бэкенд, который обрабатывает запрос неправильно в ходе трёх первых попыток обратиться к нему, возвращая данные лишь с четвёртой попытки.
Обычно, пользуясь Angular, мы создаём сервис, подключаем HttpClient и используем его для получения данных с бэкенда.
Тут нет ничего особенного. Мы подключаем модуль Angular HttpClient и выполняем простой GET-запрос. Если запрос вернёт ошибку — мы выполняем некий код для её обработки и возвращаем пустой Observable (наблюдаемый объект) для того чтобы проинформировать об этом то, что инициировало запрос. Этот код как бы говорит: «Возникла ошибка, но всё в порядке, я с этим справлюсь».
Большинство приложений выполняет HTTP-запросы именно так. В вышеприведённом коде запрос выполняется лишь один раз. После этого он либо возвращает данные, полученные с сервера, либо оказывается неудачным.
Как повторить запрос в том случае, если конечная точка /greet недоступна или возвращает ошибку? Может быть, существует подходящий оператор RxJS? Конечно, он существует. В RxJS есть операторы для всего чего угодно.
Максимальное число повторных подписок ограничено count (это — числовой параметр, передаваемый методу)».
Оператор retry очень похож на то, что нам нужно. Поэтому давайте встроим его в нашу цепочку.
Как уже было сказано, бэкенд возвращает ошибку при выполнении первых трёх попыток выполнить к нему запрос, а при поступлении к нему четвёртого запроса возвращает обычный ответ.
На вкладке инструментов разработчика Network можно видеть, что оператор retry решает поставленную перед ним задачу и трижды повторяет выполнение неудавшегося запроса. Последняя попытка оказывается удачной, приложение получает ответ, на странице появляется соответствующее сообщение.
Всё это очень хорошо. Теперь приложение умеет повторять неудавшиеся запросы.
Однако этот пример ещё можно улучшать. Обратите внимание на то, что сейчас повторные запросы выполняются немедленно после выполнения запросов, которые оказываются неудачными. Такое поведение системы не принесёт особой пользы в нашей ситуации — тогда, когда поезд попадает в туннель и интернет-соединение на некоторое время теряется.
Отложенный повтор неудачных запросов
Поезд, который попал в туннель, не выезжает из него мгновенно. Он проводит там некоторое время. Поэтому нужно «растянуть» период, в течение которого мы выполняем повторные запросы к серверу. Сделать это можно, отложив выполнение повторных попыток.
Для этого нам нужно лучше контролировать процесс выполнения повторных запросов. Нам нужно, чтобы мы могли бы принимать решения о том, когда именно нужно выполнять повторные запросы. Это значит, что возможностей оператора retry нам уже недостаточно. Поэтому снова обратимся к документации по RxJS.
Да уж, определение не из простых. Давайте опишем то же самое более доступным языком.
▍Создание собственного оператора delayedRetry
Мы можем решить проблему управления состоянием и улучшить читабельность кода, оформив вышеприведённый код в виде отдельного оператора RxJS.
Существуют разные способы создания собственных RxJS-операторов. То, каким именно методом воспользоваться, сильно зависит от того, как устроен тот или иной оператор.
Наш оператор построен на базе существующих операторов RxJS. В результате мы можем воспользоваться самым простым способом создания собственных операторов. В нашем случае RxJs-оператор — это всего лишь функция со следующей сигнатурой:
Этот оператор принимает исходный Observable и возвращает ещё один Observable.
Так как наш оператор позволяет пользователю указать то, как часто должны выполняться повторные запросы, и то, сколько раз их нужно выполнять, нам нужно обернуть вышеприведённое объявление функции в фабричную функцию, которая принимает в качестве параметров значения delayMs (задержка между повторами) и maxRetry (максимальное количество повторов).
Если вам это интересно — загляните сюда.
Итак, давайте, основываясь на вышеприведённых фрагментах кода, напишем собственный RxJs-оператор.
Отлично. Теперь мы можем импортировать этот оператор в клиентский код. Воспользуемся им при выполнении HTTP-запроса.
Мы поместили оператор delayedRetry в цепочку и передали ему, в качестве параметров, числа 1000 и 3. Первый параметр задаёт задержку в миллисекундах между попытками выполнения повторных запросов. Второй параметр определяет максимальное количество повторных запросов.
Перезапустим приложение и посмотрим на то, как работает новый оператор.
Проанализировав поведение программы с помощью инструментов разработчика браузера, мы можем увидеть, что выполнение повторных попыток выполнения запроса откладывается на секунду. После получения правильного ответа на запрос в окне приложения появится соответствующее сообщение.
Экспоненциальное откладывание запроса
Давайте разовьём идею отложенного повтора неудачных запросов. Ранее мы всегда откладывали выполнение каждого из повторных запросов на одно и то же время.
Здесь же мы поговорим о том, как увеличивать задержку после каждой попытки. Первая попытка повтора запроса выполняется через секунду, вторая — через две секунды, третья — через три.
Если воспользоваться этим оператором в приложении и испытать его — можно увидеть, как задержка выполнения повторного запроса растёт после каждой новой попытки.
После каждой попытки мы определённое время выжидаем, повторяем запрос и увеличиваем время ожидания. Тут, как обычно, после того, как сервер вернёт правильный ответ на запрос, мы выводим сообщение в окне приложения.
Итоги
Повтор неудачных HTTP-запросов делает приложения стабильнее. Это особенно значимо при выполнении очень важных запросов, без данных, получаемых посредством которых, приложение не может нормально работать. Например, это могут быть конфигурационные данные, содержащие адреса серверов, с которыми нужно взаимодействовать приложению.
В большинстве сценариев оператора RxJs retry недостаточно для организации надёжной системы повтора неудачных запросов. Оператор retryWhen даёт разработчику более высокий уровень контроля над повторными запросами. Он позволяет настраивать интервал выполнения повторных запросов. Благодаря возможностям этого оператора можно реализовать схему отложенных повторов или повторов с экспоненциальным откладыванием.
При реализации в цепочках RxJS шаблонов поведения, которые подходят для повторного использования, рекомендуется оформлять их в виде новых операторов. Вот репозиторий, код из которого использовался в этом материале.
Уважаемые читатели! Как вы решаете задачу повтора неудачных HTTP-запросов?
RxSwift: немного о share(), replay(), shareReplayLatestWhileConnected() и других классных операторах
Я уже писал про Publish, Connect и RefCount в RxSwift. Для того, чтобы лучше раскрыть тему, представляю вашему вниманию перевод другой замечательной статьи, про различия между такими операторами, как share(), replay(), replayAll(), shareReplay(), publish() и shareReplayLatestWhileConnected().
Частая ошибка, которую совершают новички, взявшиеся за освоение Rx — это непонимание того, что цепочка операторов на Observable выполняется заново с каждым новым подписчиком:
Мы имеем несколько подписчиков на один-единственный Observable, но мы не хотим, чтобы его код исполнялся с каждым новым Subscriber’ом. Для этого в RxSwift имеется несколько операторов. Вот резюмирующая табличка, описывающая каждый из них:
1 — ретранслирует произведенных до подписки элементов не больше, чем bufferSize.
2 — ретранслирует 1 элемент, произведенный до подписки, до тех пор, пока существует хотя бы один подписчик.
Теперь рассмотрим подробно каждое свойство
Shared subscription. Возвращаемый Observable делит одну основную подписку между всеми подписчиками.
Reference counting. Возвращаемый Observable считает количество сабскрайберов, которые на него подписаны. Как мы помним из первого свойства, на Observable, к которому был применен оператор (его будем называть Source Observable), всегда существует только одна «основная» подписка и она делится данными с остальными подписчиками.
Когда счетчик подписчиков достигает нуля, происходит очистка ресурсов Source Observable. Обратите внимание: каждый раз, когда счетчик увеличивается с нуля до единицы, на Source Observable будет производиться новая подписка. Но этого может не случиться, если Source Observable до этого завершился или послал ошибку. Это довольно непредсказуемое поведение, поэтому я стараюсь избегать его и при написании кода убеждаюсь, что после того, как количество подписчиков опустилось до нуля, не будет никаких последующих подписок.
Изучаем мультикаст операторы RxJS
Привет, Хабр! Представляю вашему вниманию перевод статьи «Understanding RxJS Multicast Operators» автора Netanel Basal.
Широковещательные или мультикаст операторы нередко кажутся самой сложной темой при изучении RxJS. В этой статье я попробую все доступно объяснить.
Мы рассмотрим внутреннее устройство мультикаст операторов и решаемые ими задачи.
Давайте начнем с описания основных строительных блоков RxJS.
Observable
В RxJS наблюдаемые объекты (далее «потоки») изначально являются холодными. Это значит, что каждый раз при подписке на поток выполняется коллбэк подписки.
Для лучшего понимания создадим такую реализацию:
Это все довольно просто. Давайте теперь перейдем ближе к реальности. Например завернем в поток нативный XHR API
Теперь, глядя на нашу реализацию, как вы думаете, что произойдет когда мы подпишемся на этот поток дважды?
Правильно, выполнится два http-запроса. Если еще раз взглянуть на реализацию класса Observable, мы увидим почему так. Каждый подписчик вызывает коллбэк подписки, который в свою очередь каждый раз выполняет http-запрос.
Операторы
Оператор это функция, которая принимает на вход поток, производит какие-либо действия, и возвращает поток.
Напишем наш первый собственный оператор.
Т.е. внутри него обязательно присутствует подписка на входной поток.
Перед тем как использовать этот новый оператор, нам нужно как-то прикрепить его к потоку. Расширим наш класс Observable методом pipe()
Простой метод, всего одна строка кода. pipe принимает массив операторов и вызывает их по очереди, передавая каждому на вход результат выполнения предыдущего.
Давайте же используем наш оператор:
Если мы подпишемся на цепочку дважды, каждая подписка в цепочке вызовется два раза.
А если нам не подходит такое поведение? Если мы хотим вызвать функцию подписки только один раз, сколько бы у нас не было подписок?
Subjects
Давайте напишем его.
Subject может выполнять роль промежуточного звена между холодным потоком и множеством подписчиков.
Изменим наш пример следующим образом:
Теперь у нас оригинальный коллбэк подписки выполняется один раз, и только один http-запрос будет выполнен.
Подписчики, опоздавшие на вечеринку
Что произойдет, если исходный поток уже сработал до того, как мы подписались?
Не получится показать это на прошлом примере, так как http — асинхронный, даже если подписаться на него сразу после, значение все равно придет уже после подписки.
Давайте быстро создадим порождающую функцию of :
Наши подписчики не получили ничего. Почему? Наша реализация не поддерживает «опоздавших» подписчиков. Когда исходный поток от of() эмитит значения, подписчики еще не зарегистрированы, эти значения уйдут в никуда.
В актуальных примерах на Angular вполне может быть такое, что исходный поток сработал, но ваш компонент еще не присутствует на странице. И когда компонент появится, он подпишется на источник, но не получит уже пролетевших значений.
Каждое выпущенное значение будет передано всем текущим подписчикам и сохранено для будущих, размер буфера bufferSize устанавливается в конструкторе.
Несмотря на позднюю подписку мы поймали их всех.
Резюмируя, предназначение ReplaySubject — это рассылка значений всем подписчикам и кэширование их для будущих «опоздавших» подписчиков.
Теперь мы наконец переходим к мультикаст операторам. Надеюсь рассмотренные выше примеры выше вам помогут быстрее их понять.
Multicast Operators
Multicast и Connect
Метод connect позволяет нам самим определять, когда запустить на выполнение исходный поток. Тут есть момент, о котором нужно помнить — чтобы отписаться от источника нужно выполнить:
По этому коду можно догадаться, что произойдет под капотом.
Переиспользовать уже завершенный subject нельзя, фабричная функция решает эту проблему.
RefCount
Автоматизация процесса позволила бы избежать ошибок и упростила код. Добрые разработчики RxJS подумали об этом за нас и создали refCount оператор.
Publish и его варианты
multicast() + Subject + refCount() это довольно типичный случай в RxJS и разработчики сократили его до одного оператора.
Посмотрим какие у нас есть варианты.
В этом случае подсчета ссылок не будет. Это значит что пока исходный поток не будет завершен, shareReplay будет подписан на него, независимо есть у него самого конечные подписчики или нет. Все новые подписчики получат последние x значений.
shareReplay vs publishReplay + refCount
Посмотрим в чем сходство и в чем разница.
У них одинаковое поведение refCount — подписка и отписка от исходного потока на основании количества подписчиков. Еще у них одинаково реагируют когда исходный поток завершился — все новые подписчики получает X последних значений.
Примеры, иллюстрирующие это:
Актуальные примеры в Angular
Посмотрим как применять в боевых условиях изученные мультикаст операторы.
Используем share
Пусть у нас есть компонент, которому нужны данные из исходного потока. Это может быть запрос http, стейт или что-угодно. А еще нам нужно манипулирование данными, типа фильтрации, сортировки и т.п.
А теперь нам нужен другой компонент, который показывает только первого пользователя. Если мы подпишемся на исходных поток как он есть, то:
И вот у нас два http-запроса, операции сортировки или фильтрации выполнятся дважды.
Применяем share :
Используем ShareReplay
ShareReplay применяется когда нужно эмитить, кэшировать и повторять последние X значений. Типичный пример — синглтон сервис, выполняющий http запрос.