Angular 等待所有订阅完成

Question

在 Angular 中，一个页面对多个操作进行多个 http 调用，比方说按钮点击。但是当按下最后一个“完成”按钮时，我想确保所有这些请求在它继续之前完成。我尝试将 forkJoin 与可观察对象一起使用，但它本身会触发请求，这不是我想要做的，我希望其他操作触发请求，只是为了确保在单击“完成”时完成异步请求。有了承诺，我只会将承诺推送到数组，然后执行 Promise.all(allRequests).then(()=>{})

observables: Observable<any>[];

onBtn1Click(){
   let o1 = this.service.doAction1();
   this.observables.push(o1);
   
   o1.subscribe(resp => {
        //do action1
   });
}

onBtn2Click(){
   let o2 = this.service.doAction2();
   this.observables.push(o2);
   
   o2.subscribe(resp => {
        //do action2
   });
}

onDoneClick(){
   // I would like something like this just that it wouldn't trigger the requests but make sure they are completed. 
   forkJoin(this.observables).subscribe(()=>{
     //Proceed with other things
   });
}

Answer 1

除非有人想出一个优雅的方法，否则应该使用以下方法。

我正在创建一个对象来保存来自 HTTP 请求的每个冷可观察对象的热可观察对象。该请求将使用 RxJS finalize 运算符发送到相应的热可观察对象。然后可以使用 forkJoin 和 take(1) 组合这些热可观察对象以等待源请求完成。

private httpReqs: { [key: string]: ReplaySubject<boolean> } = Object.create(null);

onBtn1Click() {
  this.httpReqs['btn1'] = new ReplaySubject<boolean>(1);
  this.service.doAction1().pipe(
    finalize(() => this.httpReqs['btn1'].next(true))
  ).subscribe(resp => {
    // do action1
  });
}

onBtn2Click() {
  this.httpReqs['btn2'] = new ReplaySubject<boolean>(1);
  this.service.doAction1().pipe(
    finalize(() => this.httpReqs['btn2'].next(true))
  ).subscribe(resp => {
    // do action2
  });
}

onDoneClick(){
  forkJoin(
    Object.values(this.httpReqs).map(repSub =>
      repSub.asObservable().pipe(
        take(1)
      )
    )
  ).subscribe(() => {
    // Proceed with other things
  });
}

Answer 2

使用`shareReplay`

如果您进行多播，任何订阅已完成流的订阅者都会收到 complete 通知。你可以利用它。

各种共享运算符都有一个隐式的 refCount，每隔几个 RxJS 版本就会更改其默认值。 shareReplay(n) 的当前版本非常直观，但您可能需要在旧版本上设置 refCount:false，甚至使用 multicast(new ReplaySubject(1)), refCount()

onBtn1Click(){
  let o1 = this.service.doAction1().pipe(
    shareReplay(1)
  );
  this.observables.push(o1);
   
  o1.subscribe(resp => {
    //do action1
  });
}

这是让您的代码按您希望的方式工作的最小更改

扫码计数activity

如果只计算当前活动的操作，则可以完全避免 forkJoin。

count = (() => {
  const cc = new BehaviorSubject<number>(0);
  return {
    start: () => cc.next(1),
    stop: () => cc.next(-1),
    value$: cc.pipe(
      scan((acc, curr) => acc + curr, 0)
    )
  }
})();

onBtn1Click(){
  this.count.start();

  this.service.doAction1().pipe(
    finalize(this.count.stop)
  ).subscribe(resp => {
    //do action1
  });
}

onDoneClick(){
  this.count.value$.pipe(
    first(v => v === 0) // Wait until nothing is currently active
  ).subscribe(() => {
    //Proceed with other things
  });

}

Angular 等待所有订阅完成

Angular wait for all subscriptions to complete

javascript

angular

rxjs

observable

subscription

使用`shareReplay`

扫码计数activity

Angular 等待所有订阅完成

Angular wait for all subscriptions to complete

javascript

angular

rxjs

observable

subscription

使用shareReplay

扫码计数activity

使用`shareReplay`