RxJS : how to organize subscription betwen nested subscribing calls

Question

Within my Angular app i ve the following treatment :

• OnInit on lauching a subscibe from a subject call (SubjectOne)

• when there is a new data coming from SubjectOne ,

  and if some condition is verified ; i reeuse this data to launch a second call whitch is a http call from a service call .

Here is my code

MyComponent.ts :

  ngOnInit() {
    this.getFirstTreatment();
  }

  getFirstTreatment() {
    this.subscriptionOne = this.myService.subjectOne.subscribe((data) => {
      this.myValue = data['myValue'];
      this.getSecondTreatment(data['myValue'])
    })
  }

  getSecondTreatment(thatValue) {
    if(thatValue >= 100){
     this.subscriptionTwo = this.myService.sendToBackend(thatValue).subscribe((response)=>{}
    }
  }

MyService.ts

sendToBackend(thatValue){
    let newValue = someFormatingnMethod(thatValue)
    return this.httpClient.post(url , newValue );
}

My Purpose is how may i dynamically close the subscribtionTwo so it won't be called n times after each time i got new data from the subject .

NB : mySubject can notice some new Data even before the destroy of the compoment

I ve tried to use switchMap , but it seems to not work correctly

Suggestions ?

Answer 1

• You are starting with one observable
• That observable stays open after it has emitted a value, so we need to unsubscribe
• You then want to conditionally run a second observable based on the result of the first observable

I would take this approach:

• Set up your first observable as you are currently doing
• Use takeUntil to unsubscribe on destroy
• Use filter to only continue based on a condition
• Use switchMap to run the second observable
• The second observable is an HttpClient request, which self-completes, so we don't need to unsubscribe

private destroyed$ = new Subject();

ngOnInit() {
  getFirstTreatment();
}

ngOnDestroy() {
  this.destroyed$.next();
  this.destroyed$.complete();
}

getFirstTreatment() {
  this.myService.subjectOne.pipe(
    takeUntil(this.destroyed$),
    tap(data => this.myValue = data['myValue']),    
    filter(data => data['myValue'] >= 100),
    switchMap(data => this.getSecondTreatment(data['myValue']))
  ).subscribe(data => {
    console.log(data); // the output of the second observable    
  });
}

getSecondTreatment(myValue): Observable<any> {
  return this.getSecondTreatment(myValue);
}