Rx JS: как организовать подписку между вложенными звонками

Question

В моем приложении Angular я выполняю следующую процедуру:

• OnInit при отсоединении абонента от предметного вызова (SubjectOne)

• при появлении нового данные, поступающие из SubjectOne,

  и если какое-либо условие проверено; я повторно использую эти данные для запуска второго вызова, который является http-вызовом из сервисного вызова.

Вот мой код

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

Моя цель - как мне динамически закрыть subscriptionTwo , чтобы оно не было вызывается n раз после каждого получения новых данных от субъекта.

NB: mySubject может заметить некоторые новые данные даже до уничтожения композита

Я пытался использовать switchMap, но, похоже, он не работает правильно

Предложения?

Answer 1

• Вы начинаете с одной наблюдаемой
• Эта наблюдаемая остается открытой после получения значения, поэтому нам нужно отписаться
• Затем вы хотите условно запустить вторую наблюдаемую базу. на результат первой наблюдаемой

Я бы выбрал такой подход:

• Настройте свою первую наблюдаемую, как вы в настоящее время делаете
• Используйте takeUntil чтобы отменить подписку на уничтожение
• Используйте filter, чтобы продолжить только на основе условия
• Используйте switchMap для запуска второй наблюдаемой
• Вторая наблюдаемая - HttpClient запрос, который завершается самостоятельно, поэтому нам не нужно отписываться

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