添加到之前的结果，Observable 管道只运行一次

Question

live example

我有一个过滤器数组作为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]);
}

Answer 1

其实是因为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);
}