RxJS: Не отписывайся / Хабр

Прекратите выдавать за утечку памяти что-то другое

@Component({
    selector: 'app-bad-modal',
    templateUrl: './bad-modal.component.html',
    styleUrls: ['./bad-modal.component.css'],
    providers: [HttpClient]
})
export class BadModalComponent implements OnInit, OnDestroy {

    loading = false;
    largeData: number[] = (new Array(1000000)).fill(1);
    destroyed$ = new Subject<void>();
    data: DataInterface;

    constructor(
        private http: HttpClient,
        private bsModalRef: BsModalRef
    ) {
    }

    ngOnInit() {
        this.loadData();
    }

    loadData(): void {
        // For example only, not for production.

        this.loading = true;
        const subscription = this.http.get<DataInterface>('/action.php?a=2').pipe(
                takeUntil(this.destroyed$),
                catchError((err) => throwError(err.message)),
                finalize(() => console.log('FINALIZE'))
            )
            .subscribe({
                next: (res) => {
                    setTimeout(() => {
                        console.log(subscription.closed ? 'SUBSCRIPTION IS CLOSED' : 'SUBSCRIPTION IS NOT CLOSED!');
                    }, 0);
                    console.log('LOADED');
                    this.data = res;
                    this.loading = false;
                },
                error: (error) => {
                    setTimeout(() => {
                        console.log(subscription.closed ? 'ERROR - SUBSCRIPTION IS CLOSED' : 'ERROR - SUBSCRIPTION IS NOT CLOSED!');
                    }, 0);
                    console.log('ERROR', error);
                },
                complete: () => {
                    setTimeout(() => {
                        console.log(subscription.closed ? 'COMPLETED - SUBSCRIPTION IS CLOSED' : 'COMPLETED - SUBSCRIPTION IS NOT CLOSED!');
                    }, 0);
                    console.log('COMPLETED');
                }
            });
    }

    close(event?: MouseEvent): void {
        if (event) {
            event.preventDefault();
        }
        this.bsModalRef.hide();
    }

    ngOnDestroy() {
        console.log('DESTROY');
        this.destroyed$.next();
        this.destroyed$.complete();
    }
}

@Component({
  selector: 'app-good-modal',
  templateUrl: './good-modal.component.html',
  styleUrls: ['./good-modal.component.css']
})
export class GoodModalComponent implements OnInit, OnDestroy {

    loading = false;
    largeData: number[] = (new Array(1000000)).fill(1);
    data: DataInterface;

    constructor(
        private http: HttpClient,
        private bsModalRef: BsModalRef
    ) {
    }

    ngOnInit() {
        this.loadData();
    }

    loadData(): void {
        // For example only, not for production.

        this.loading = true;
        const subscription = this.http.get<DataInterface>('/action.php?a=1').pipe(
            catchError((err) => throwError(err.message)),
            finalize(() => console.log('FINALIZE'))
        )
            .subscribe({
                next: (res) => {
                    setTimeout(() => {
                        console.log(subscription.closed ? 'SUBSCRIPTION IS CLOSED' : 'SUBSCRIPTION IS NOT CLOSED!');
                    }, 0);
                    console.log('LOADED');
                    this.data = res;
                    this.loading = false;
                },
                error: (error) => {
                    setTimeout(() => {
                        console.log(subscription.closed ? 'ERROR - SUBSCRIPTION IS CLOSED' : 'ERROR - SUBSCRIPTION IS NOT CLOSED!');
                    }, 0);
                    console.log('ERROR', error);
                },
                complete: () => {
                    setTimeout(() => {
                        console.log(subscription.closed ? 'COMPLETED - SUBSCRIPTION IS CLOSED' : 'COMPLETED - SUBSCRIPTION IS NOT CLOSED!');
                    }, 0);
                    console.log('COMPLETED');
                }
            });
    }

    close(event?: MouseEvent): void {
        if (event) {
            event.preventDefault();
        }
        this.bsModalRef.hide();
    }

    ngOnDestroy() {
        console.log('DESTROY');
    }
}

— изменить 2 (2022/12/28)

Источник 5

В учебнике Angular глава Routing теперь заявляет следующее: «Маршрутизатор управляет наблюдаемыми объектами, которые он предоставляет, и локализует подписки. Подписки очищаются при уничтожении компонента, защищая от утечек памяти, поэтому нам не нужно отписываться от параметры маршрута наблюдаемые. ” – Марк Райкок

Вот обсуждение вопросов Github для Angular docs, касающихся Router Observables, где Уорд Белл упоминает, что прояснение всего этого находится в работе.

— оригинальный ответ

TLDR:

Для этого вопроса существует (2) вида Observables- конечное значение и бесконечное значение.

— редактировать 1

Источник 4

— редактировать 3 – «официальное» решение (2022/04/09)

Я говорил с Уордом Беллом об этом вопросе в NGConf (я даже показал ему этот ответ, который, по его словам, был правильным), но он сказал мне, что команда разработчиков документации для Angular нашла решение этого вопроса, которое не было опубликовано (хотя они работают над его утверждением). ). Он также сказал мне, что я могу обновить свой SO-ответ следующей официальной рекомендацией.

Решение, которое мы все должны использовать в будущем, заключается в добавлении private ngUnsubscribe = new Subject();поля ко всем компонентам, к которым .subscribe()обращаютсяObservable s в своем коде класса.

— редактировать 4 – дополнительные ресурсы (2022/09/01)

В недавнем эпизоде приключений в Angular Бен Леш и Уорд Белл обсуждают вопросы о том, как и когда отписаться в компоненте. Обсуждение начинается примерно в 1:05:30.

Javascript – отписаться от rxjs observables –

   ConnectableObservable {
     source: EventPatternObservable { _add: [Function], _del: [Function], _fn: undefined },
     _connection: ConnectDisposable { _p: [Circular], _s: [Object] },
     _source: AnonymousObservable { source: [Object], __subscribe: [Function: subscribe] },
     _subject: 
      Subject {
        isDisposed: false,
        isStopped: false,
        observers: [Object],
        hasError: false } },
  _count: 1,
  _connectableSubscription: 
   ConnectDisposable {
     _p: 
      ConnectableObservable {
        source: [Object],
        _connection: [Circular],
        _source: [Object],
        _subject: [Object] },
     _s: AutoDetachObserver { isStopped: false, observer: [Object], m: [Object] } } }

FilterObservable {
  source: 
   RefCountObservable {
     source: 
      ConnectableObservable {
        source: [Object],
        _connection: [Object],
        _source: [Object],
        _subject: [Object] },
     _count: 1,
     _connectableSubscription: ConnectDisposable { _p: [Object], _s: [Object] } },
  predicate: [Function] }

'use strict';
var rx = require('rx');

function normalizeKeypressEvents(value, key) {
  return {value: value, key: key || {}};
}

module.exports = function (rl) {

  var keypress = rx.Observable.fromEvent(rl.input, 'keypress', normalizeKeypressEvents)
    .filter(function (e) {
      // Ignore `enter` key. On the readline, we only care about the `line` event.
      return e.key.name !== 'enter' && e.key.name !== 'return';
    });

  return {
    line: rx.Observable.fromEvent(rl, 'line'),

    keypress: keypress,

    normalizedLeftKey: keypress.filter(function (e) {
      return e.key.name === 'left';
    }).share(),

    normalizedRightKey: keypress.filter(function (e) {
      return e.key.name === 'right';
    }).share(),

    normalizedUpKey: keypress.filter(function (e) {
      return e.key.name === 'up' || e.key.name === 'k' || (e.key.name === 'p' && e.key.ctrl);
    }).share(),

    normalizedDownKey: keypress.filter(function (e) {
      return e.key.name === 'down' || e.key.name === 'j' || (e.key.name === 'n' && e.key.ctrl);
    }).share(),

    numberKey: keypress.filter(function (e) {
      return e.value && '123456789'.indexOf(e.value) >= 0;
    }).map(function (e) {
      return Number(e.value);
    }).share(),

    spaceKey: keypress.filter(function (e) {
      return e.key && e.key.name === 'space';
    }).share(),

    aKey: keypress.filter(function (e) {
      return e.key && e.key.name === 'a';
    }).share(),

    iKey: keypress.filter(function (e) {
      return e.key && e.key.name === 'i';
    }).share()
  };
};

var source = Rx.Observable.fromEvent(input, 'click');

var subscription = source.subscribe(
  function (x) {
    console.log('Next: Clicked!');
  },
  function (err) {
    console.log('Error: %s', err);
  },
  function () {
    console.log('Completed');
  });

events.normalizedUpKey.takeUntil(validation.success).forEach(this.onUpKey.bind(this));
events.normalizedDownKey.takeUntil(validation.success).forEach(this.onDownKey.bind(this));

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

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

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

Загадка на посошок

Если у вас есть желание немного поупражняться, решите предыдущую задачу, но для создания продукта. То есть сначала создаём продукт, потом обновляем теги, а только потом — категории.

Использование observable для автокомплита или поиска

Задача: Показывать предложения страниц при вводе данных на форме

Решение: Подпишемся на изменение данных формы, возьмём только меняющиеся данные инпута, поставим небольшую задержку, чтобы событий не было слишком много и отправим запрос в википедию. Результат выведем в консоль. Интересный момент в том, что switchMap отменит предыдущий запрос, если пришли новые данные.

Использование takeuntil для отписки

Отписываться можно и более изящным вариантом, особенно если в компоненте присутствует больше двух подписок:

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

Возьмем, к примеру, этот выдуманный компонент (это специально не 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, что является некоторым порогом, на котором я хочу остановить поток данных. (Не знаю, зачем, это просто странный компонент).

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

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

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

Кеширование запроса

Задача: Необходимо закешировать Observable запрос

Остановка анимации загрузки после окончания выполнения подписки

Задача: Показать значок загрузки после начала сохранения данных и скрыть его, когда данные сохранятся или произойдет ошибка.

Решение: За отображение загрузчика у нас отвечает переменная loading, после нажатия на кнопку, установим ее в true. А для установки ее в false воспользуемся Observable.finally функций, которая выполняется после завершения подписки или если произошла ошибка.

Последовательный combinelatest

Задача: Критическая ситуация на сервере! Backend команда сообщила, что для корректного обновления продукта нужно выполнять строго последовательно:

1. Обновление данных продукта (заголовок и описание);
2. Обновление списка тегов продукта;
3. Обновление списка категорий продукта;

Решение: У нас есть 3 Observable, полученных из productService. Воспользуемся concatMap:

const updateProduct$ = this.productService.update(product);
const updateTags$ = this.productService.updateTags(productId, tagList);
const updateCategories$ = this.productService.updateCategories(productId, categoryList);

Observable
  .from([updateProduct$, updateTags$, updateCategories$])
  .concatMap(a => a)  // выполняем обновление последовательно
  .toArray()          // Возвращает массив из последовательности
  .subscribe(res => console.log(res)); // res содержит массив результатов запросов

Резюме: используйте takeuntil, takewhile и пр.

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

Скомпонуйте управление подписками с помощью 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` на возвращенном объекте-подписке, вы не будете уведомлены о том, что подписка была отменена.

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

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

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

Создание собственного источника событий

Задача: Создать переменную lang$ в configService, на которую другие компоненты будут подписываться и реагировать, когда язык будет меняться.

Решение: Воспользуемся классом BehaviorSubject для создания переменной lang$;

ОтличияBehaviorSubject от Subject:

1. BehaviorSubject должен инициализироваться с начальным значением;
2. Подписка возвращает последнее значение Subjectа;
3. Можно получить последнее значение напрямую через функцию getValue().

Создаём переменную lang$ и сразу инициализируем. Так же добавляем функцию setLang для установки языка.

// configService
lang$: BehaviorSubject<Language> = new BehaviorSubject<Language>(DEFAULT_LANG);
setLang(lang: Language) {
  this.lang$.next(this.currentLang); // тут мы поставим
}

Подписываеся на изменение языка в компоненте. Переменная lang$ является “горячим” Observable объектом, то есть подписка требует отписки при разрушении объекта.

private subscriptions: Subscription[] = [];
ngOnInit() {
  const langSub = this.configService.lang$
    .subscribe(() => {
      // ...
    });
  this.subscriptions.push(langSub);
}
ngOnDestroy() {
  this.subscriptions
    .forEach(s => s.unsubscribe());
}