In Angular a page makes multiple http calls on multiple actions, let's say button clicks. But when the last "DONE" button is pressed I want to make sure that all those requests are finished before it progresses. I tried to use forkJoin with observables but it triggers requests itself which is not what I want to do, I want other actions to trigger requests and just to make sure that async requests are finished when "DONE" is clicked. With promises I would just push promises to array and then do 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 pleted. 
   forkJoin(this.observables).subscribe(()=>{
     //Proceed with other things
   });
}

In Angular a page makes multiple http calls on multiple actions, let's say button clicks. But when the last "DONE" button is pressed I want to make sure that all those requests are finished before it progresses. I tried to use forkJoin with observables but it triggers requests itself which is not what I want to do, I want other actions to trigger requests and just to make sure that async requests are finished when "DONE" is clicked. With promises I would just push promises to array and then do 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 pleted. 
   forkJoin(this.observables).subscribe(()=>{
     //Proceed with other things
   });
}

• javascript
• angular
• rxjs
• observable
• subscription

• Take a look at finalize() operator in Rxjs – Kevin Zhang Commented Jan 8, 2021 at 10:39

2 Answers 2

Unless someone es up with an elegant approach, the following should do it.

I'm creating an object to hold hot observable for each cold observable from the HTTP request. The request would emit to it's corresponding hot observable using RxJS finalize operator. These hot observables could then be bined using forkJoin with a take(1) to wait for the source requests to plete.

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
  });
}

Using shareReplay

If you multicast, any subscriber who subscribes to a pleted stream gets the plete notification. You can leverage that.

The various share operators have an implicit refCount that changes its default every few RxJS versions. The current version for shareReplay(n) is pretty intuitive, but you may need to set refCount:false on older versions, or even use multicast(new ReplaySubject(1)), refCount()

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

This is the smallest change that should get your code working the way you'd like

Scan to count activity

You can avoid forkJoin entirely if you just count currently active operations.

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
  });

}