RxJS: Не отписывайся

Привет, Хабр! Представляю вашему вниманию перевод статьи "RxJS: Don’t Unsubscribe" автора Ben Lesh.

Ну… ладно, просто не отказывайся от подписок.

Я часто помогаю кому-нибудь в отладке проблем с их RxJS кодом, в том числе со структурированием приложений, которые несут в себе много асинхронного кода. При этом я всегда вижу одно и тоже, как люди держат обработчики на тоннах подписок. Разработчик делает 3 HTTP-запроса с Observable, сохраняя 3 объекта подписки, которые будут вызваны, когда произойдет какое-то событие.

Я знаю, почему так происходит. Люди привыкли использовать `addEventListener` N раз, а затем, когда они больше не нужны, вызывать `removeEventListener` N раз. Естественным будет делать то же самое и с объектами-подписками, и по большей части вы будете правы. Но есть и лучшие способы. Сохранение слишком большого количества объектов подписок — это знак того, что вы управляете своими подписками императивно и не пользуетесь преимуществами Rx.

Как выглядит императивное управление подпиской

Возьмем, к примеру, этот выдуманный компонент (это специально не React и не Angular, а просто общий пример):

class MyGenericComponent extends SomeFrameworkComponent {
 updateData(data) {
  // что-нибудь специфичное для обновления компонента
 }

 onMount() {
  this.dataSub = this.getData()
   .subscribe(data => this.updateData(data));

  const cancelBtn = this.element.querySelector(‘.cancel-button’);
  const rangeSelector = this.element.querySelector(‘.rangeSelector’);

  this.cancelSub = Observable.fromEvent(cancelBtn, ‘click’)
   .subscribe(() => {
    this.dataSub.unsubscribe();
   });

  this.rangeSub = Observable.fromEvent(rangeSelector, ‘change’)
   .map(e => e.target.value)
   .subscribe((value) => {
    if (+value > 500) {
      this.dataSub.unsubscribe();
    }
   });
 }

 onUnmount() {
  this.dataSub.unsubscribe();
  this.cancelSub.unsubscribe();
  this.rangeSub.unsubscribe();
 }
}

В приведенном выше примере вы можете увидеть, что я вручную вызываю `unsubscribe` на трех объектах подписки в методе `onUnmount()`. Также я вызываю `this.dataSub.unsubscribe()`, когда пользователь нажимает кнопку отмены в строках 15 и 22, или когда он устанавливает селектор диапазона выше 500, что является некоторым порогом, на котором я хочу остановить поток данных. (Не знаю, зачем, это просто странный компонент).

Недостаток этого подхода в том, что я вручную управляю отменой подписки в нескольких местах в этом довольно тривиальном примере.

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

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

Скомпонуйте управление подписками с помощью takeUntil

Теперь давайте сделаем тот же пример, но используя оператор `takeUntil` из RxJS:

class MyGenericComponent extends SomeFrameworkComponent {
 updateData(data) {
  // do something framework-specific to update your component here.
 }

 onMount() {
   const data$ = this.getData();
   const cancelBtn = this.element.querySelector(‘.cancel-button’);
   const rangeSelector = this.element.querySelector(‘.rangeSelector’);
   const cancel$ = Observable.fromEvent(cancelBtn, 'click');
   const range$ = Observable.fromEvent(rangeSelector, 'change')
                            .map(e => e.target.value);
   
   const stop$ = Observable.merge(cancel$, range$.filter(x => x > 500))
   this.subscription = data$.takeUntil(stop$)
                            .subscribe(data => this.updateData(data));
 }

 onUnmount() {
  this.subscription.unsubscribe();
 }
}

Первое, что вы могли заметить, это меньший объем кода. Но это лишь одно преимущество. Еще одна вещь, которая произошла здесь, заключается в том, что я скомпоновал в поток `stop$` события, которые останавливают поток данных. Это означает, что как только я решу, что хочу добавить еще одно условие, чтобы остановить поток, например по таймеру, я могу просто добавить новый наблюдаемый объект в `stop$`. Следующая очевидная вещь — у меня есть только один объект подписки, которым я управляю императивно. Этого не изменишь, так как здесь функциональное программирование пересекается с объектно-ориентированным миром. Javascript — это язык императивный и нам приходится приходится принимать остальной мир в каком-то смысле наполовину.

Другим преимуществом этого подхода является то, что он, фактически, завершает наблюдаемый объект. Это означает, что возникнет событие завершения, которое можно обрабатать в любое время. Если вы просто вызываете `unsubscribe` на возвращенном объекте-подписке, вы не будете уведомлены о том, что подписка была отменена. Однако, если вы используете `takeUntil` (или другие операторы, перечисленные ниже), вам будет сообщено через обработчик завершения, что объект observable остановился.

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

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

Также будет очень небольшой разница в производительности между этим и простым императивным вызовом `unsubscribe`. Однако маловероятно, что она будет заметна в большинстве приложений.

Другие операторы

Есть много других способов остановить поток в «Rx-way». Я бы рекомендовал глянуть следующие операторы как минимум:

take(n): берет N значений перед остановкой наблюдаемого.
takeWhile(предикат): проверяет пропускаемые через себя значения на предикат, если он возвращает ложь, поток будет завершен.
first(): пропускает первое значение и завершает работу.
first(предикат): проверяет каждое значение на функцию предиката, если он возвращает истину, поток пропускает значение и завершается.

Резюме: Используйте takeUntil, takeWhile и пр.

Вероятно, вы должны использовать такие операторы, как `takeUntil`, чтобы управлять подписками в RxJS. Как правило, если вы видите, что есть две или больше подписки в одном компоненте, вы должны задаться вопросом, можете ли вы определить их лучше:

• во-первых, так продвинутее
• запускает событие завершения, когда вы убиваете свой поток
• обычно это меньший объем кода
• управлять становится проще
• меньше фактических точек подписки (поскольку меньше вызовов `subscribe`)

Хотите больше узнать о RxJS от автора? Заходи на rxworkshop.com!