Добавление к предыдущему результату: наблюдаемый конвейер работает только один раз

живой пример

У меня есть массив фильтров как Observable, и я хочу добавить/удалить из него фильтры. Вот код, который у меня есть, который в настоящее время добавляет Filter только при первом запуске функции.

Второй раз ничего не происходит.

private _filters$ = new BehaviorSubject<Filter[]>([]);

addFilter(added: Filter) {
    debugger
    // adding to array of filters
    this._filters$.pipe(
        tap(d => { debugger; }),
        first(), 
        map(filters => ([...filters, added]))
    ).subscribe(this._filters$);
}

Итак, мой вопрос: почему это происходит? Почему он запускается только один раз? (кстати first() не причина).

Я знаю, что могу заставить код работать так:

private _filters$ = new BehaviorSubject<Filter[]>([]);

currentFilters;

init() {
   this._filters$.subscribe(f => this.currentFilters = f);
}

addFilter(added: Filter) {
    this._filters$.next([...this.currentFilters, added]);
}

rxjs rxjs5 rxjs6
person Ced    schedule 23.06.2018    source источник

Ответы (1)


arrow_upward
2
arrow_downward

На самом деле, это из-за first. Когда вы запускаете функцию в первый раз, она создает поток и подписывается на BehaviorSubject. Когда он получает первое событие, он пересылает его на BehaviorSubject, а затем завершает BehaviorSubject. Когда вы запускаете его во второй раз, BehaviorSubject уже выключен, поэтому он немедленно отменяет все новые подписки на него.

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

// You don't actually need the caching behavior yet so just use a `Subject`
private _filters$ = new Subject<Filter>()

// Hook this up to whatever is going to be using these filters
private _pipeline$ = this._filters.pipe(
  // Use scan instead mapping back into self
  scan((filters, newFilter) => ([...filters, newFilter]), []),
  // Store the latest value for new subscribers
  shareReplay(1)
);

// Now this method is just pushing into the `Subject` and the pipeline never has to be torn down
addFilter(added: Filter) {
    debugger
    this._filters$.next(added);
}
person paulpdaniels    schedule 23.06.2018
comment
Я должен был дважды проверить first, возможно, у меня был другой, который заблокировал его в моем реальном коде. Что касается вашего ответа, я вижу одну проблему и один вопрос. Проблема в том, что _pipeline$ — это то, на что мы подписываемся, но предполагается, что мы только добавляем фильтры. Хотя в моем случае я удаляю/добавляю. Вопрос в том, использовали ли вы shareReplay(1) вместо BehaviorSubject вверху, чтобы избежать дополнительных вычислений при подписке? - person Ced; 24.06.2018
comment
Для проблемы, о которой вы упомянули, я бы предложил вам взглянуть на другой вопрос SO, который имеет аналогичную проблему (работает с числами, а не с массивами, но принцип будет таким же) stackoverflow.com/questions/31434606/. Я переключил логику на использование shareReplay, потому что это более семантично и менее подвержено ошибкам. Использование необработанных субъектов для управления состоянием обычно считается анти-паттерном, потому что чаще всего вы в конечном итоге будете терять состояние, как вы в конечном итоге обнаружили. Операторы склонны управлять собственным жизненным циклом. - person paulpdaniels; 24.06.2018
comment
Хорошо, спасибо, это выглядело довольно сложно для моего варианта использования, поэтому в итоге я использовал то, что указал в своем фрагменте, в своем вопросе. Любое встречное указание на использование этого вместо этого? - person Ced; 24.06.2018