RxJS:使用merge动态添加和删除可观察对象
我正在尝试构建一个可插拔的ProgressService。
服务应该跟踪当前正在加载的“内容”的数量。并且有一个isLoading()方法，它返回一个Observable<boolean>来指示是否正在加载任何东西。
我的第一个解决方案非常天真，使用一个new BehaviorSubject(0)，然后每个加载提供程序只调用ProgressService.increase()和ProgressService.decrease()。这样做效果很好。但现在，如果可能的话，我想做更多的反应。
然后我遇到了merge，当所有加载提供者在一开始都是已知的时候，它工作得很好：
this.progress = merge(...progressProviders).pipe(
scan((acc, curr) => acc + (curr ? 1 : -1), 0)
);当任何加载提供程序发出true或false时，这只会增加/减少progress的值。
但我还需要一些注册/注销功能。这基本上应该在merge链中添加一个新的Observable (或者删除它)。
新的ProgressService应该如下所示：
class ProgressService {
progress: Observable<number> = EMPTY; // value > 0 when something is loading
addLoadingProvider(lp: Observable<boolean>) {
// increment this.progress, when lp emits true
// decrease this.progress, when lp emits false
// nice to have: ignore if lp's first emitted value is false
}
removeLoadingProvider(lp: Observable<boolean>) {
// stop listening to lp
// clean up: decrease this.progress, if lp last emitted true
}
isLoading(): Observable<boolean> {
return this.progress.pipe(
map(val => val !== 0)
);
}
}如果我们从addLoadingProvider返回一个Subscription，并使用Subscription.unsubscribe()注销，那么可能根本不需要removeLoadingProvider方法(?)。
希望有人能告诉我如何按需merge和unmerge额外的观察值。
回答 1
Stack Overflow用户
发布于 2019-06-10 00:23:08
根据您的解释，我可以理解以下内容，如果我的理解不正确，请纠正我-
您希望收集发出布尔值的各种可观察对象，并希望以这样的方式进行跟踪:如果它们中的任何一个至少有一个可观察对象，则最终的可观察对象应发出"true"，否则最终的可观察对象应返回"false"
尽管您的BehaviorSubject方法是一种反应式方法。我提出了以下方法；让我知道它在您的场景中是否有意义-
方法1 -
enum ListAction {
Added,
Removed,
Empty,
Undefined
}
export class ProgressService {
constructor() { }
progress: Observable<number> = EMPTY; // value > 0 when something is loading
obsListChanged: BehaviorSubject<ListAction> = new BehaviorSubject<any>(ListAction.Undefined);
obsList: Array<Observable<boolean>> = [];
addLoadingProvider(lp: Observable<boolean>) {
// increment this.progress, when lp emits true
// decrease this.progress, when lp emits false
// nice to have: ignore if lp's first emitted value is false
this.obsList.push(lp);
this.obsListChanged.next(ListAction.Added);
}
removeLoadingProvider(lp: Observable<boolean>) {
// stop listening to lp
// clean up: decrease this.progress, if lp last emitted true
this.obsList = this.obsList.filter(i => i !== lp);
this.obsListChanged.next(ListAction.Removed);
}
isLoading(): Observable<boolean> {
// return this.progress.pipe(
// map(val => val !== 0)
// );
return this.obsListChanged.pipe(
switchMap(() => {
return combineLatest(this.obsList);
}),
map(v => {
return v.some(loading => loading);
})
);
}
}我已经定义了一个rxjs observable，如果你想根据你的ListAction来做特定的工作，那么你可以按照你的逻辑在ListAction运算符中做同样的事情。
方法2方法1的一个改进版本- -
export class ProgressService {
constructor() { }
progress: Observable<number> = EMPTY; // value > 0 when something is loading
obsList$: BehaviorSubject<Array<Observable<boolean>>> = new BehaviorSubject<Array<Observable<boolean>>>([]);
addLoadingProvider(lp: Observable<boolean>) {
// increment this.progress, when lp emits true
// decrease this.progress, when lp emits false
// nice to have: ignore if lp's first emitted value is false
this.obsList$.next([...this.obsList$.getValue(), lp]);
}
removeLoadingProvider(lp: Observable<boolean>) {
// stop listening to lp
// clean up: decrease this.progress, if lp last emitted true
const removed = this.obsList$.getValue().filter(i => i !== lp);
this.obsList$.next(removed);
}
isLoading(): Observable<boolean> {
// return this.progress.pipe(
// map(val => val !== 0)
// );
return this.obsList$.pipe(
switchMap(obs => {
return combineLatest(obs);
}),
map(v => {
return v.some(loading => loading);
})
);
}
}https://stackoverflow.com/questions/56513781
复制相似问题
腾讯云开发者
