RXJS - 3 параллельных наблюдаемых, затем выдать значение и использовать его для последовательного вызова 3 наблюдаемых

Question

ЗАКРЫТО

Возможно, мой вопрос немного сложен, но я не могу понять, как реализовать то, что я хочу.

Контекст
Три параллельных наблюдаемых генерируют значение, затем, когда у меня есть все три значения, я немного их изменяю и затем хочу вызвать три наблюдаемых последовательно.Как на картинке ниже:

Сейчас?
На данный момент мне удалось сделать это, поместив мои три параллельные наблюдаемые в zip оператор, затем подпишитесь на него, измените значение и по завершению, вызовите другой, подпишитесь и по завершению .. Три раза!

this.service.Function(id) //return zip(Ob1, Ob2, Ob3)
  .subscribe(
    ([val1, val2, val3]) => {
      /*DO SOMETHING*/
      this.tmp1 = val1;
      this.tmp2 = val2;
      this.tmp3 = val3;
    },
    () => {}, //on error
    () => { //on complete
      let newV1, newV2, newV3 = [];
      [newV1, newV2, newV3 ] = [
        this.tmp1.map(x => x.id2),
        this.tmp2.map(x => x.id2),
        this.tmp3.map(x => x.id2)
      ];
      this.service.Function2(newV1)
        .subscribe(res => {
            //DO SOMETHING
          },
          () => {},
          () => { //on complete
            this.service.Function2(newV2)
              .subscribe(res => {
                  //DO SOMETHING
                },
                () => {},
                () => { //on complete
                  this.service.Function2(newV3)
                    .subscribe(res => {
                        //DO SOMETHING
                      },
                      () => {},
                      () => {
                        //DO SOMETHING
                      });
                });
          });
    }
  );

То, что я пробовал
Я пробовал что-то другое с switchMap и concat , но concat не возвращает мне мои значенияas Array ...

this.kycService.getDocuments(idNotif).pipe(
  switchMap(([val1, val2, val3]) => {
    this.tmp1 = val1;
    this.tmp2 = val2;
    this.tmp3 = val3;

    let newV1, newV2, newV3 = [];
      [newV1, newV2, newV3 ] = [
        this.tmp1.map(x => x.id2),
        this.tmp2.map(x => x.id2),
        this.tmp3.map(x => x.id2)
      ];
    return concat(this.service.Function2(newV1),this.service.Function2(newV2), this.service.Function2(newV3))
  }))
  .subscribe(([Ob_newV1, Ob_newV2, Ob_newV3]) => {
    //DO SOMETHING
    //[Ob_newV1, Ob_newV2, Ob_newV3] Doesn't work, I need to do val => {}
  })
);

Если у вас есть какие-либо советы по поводу того, что использовать, меня немного смущают все операторы / функции в RXJS ..

Заранее спасибо

Мое решение

    this.kycService.getDocuments(idNotif).pipe(
  switchMap(([val1, val2, val3]) => {
    this.tmp1 = val1;
    this.tmp2 = val2;
    this.tmp3 = val3;

    let newV1, newV2, newV3 = [];
      [newV1, newV2, newV3 ] = [
        this.tmp1.map(x => x.id2),
        this.tmp2.map(x => x.id2),
        this.tmp3.map(x => x.id2)
      ];
    return concat(this.service.Function2(newV1),this.service.Function2(newV2), this.service.Function2(newV3))
  }))
  .subscribe((val) => {
    //DO SOMETHING
    //val is emitted every time my function2 complete, so I manage to deal with this and rearrange my data
  })
);

Answer 1

Зависит от ваших наблюдаемых типов.

Для горячих наблюдаемых объектов (предметов, магазинов и т. Д.) Вы будете использовать combineLatest.

Для холодных наблюдаемых (из, HTTP-вызовов, из обещаний и т. Д.) Вы будете использовать forkJoin.

Давайте предположим, что им холодно.

forkJoin(
  first$.pipe(
    map(result => /* transformation of your first observable */),
    switchMap(result => this.myService.getNextObservableFromFirst())
  ),
  second$.pipe(
    map(result => /* transformation of your second observable */),
    switchMap(result => this.myService.getNextObservableFromSecond())
  ),
  third$.pipe(
    map(result => /* transformation of your third observable */),
    switchMap(result => this.myService.getNextObservableFromThird())
  ),

).subscribe([r1, r2, r3] => /* What to do once all calls are completed */);

Кажется, что адаптированные к вашему случаю, это даст

forkJoin(
  first$.pipe(
    map(result => result.id2),
    switchMap(result => this.myService.Function2(result))
  ),
  second$.pipe(
    map(result => result.id2),
    switchMap(result => this.myService.Function2(result))
  ),
  third$.pipe(
    map(result => result.id2),
    switchMap(result => this.myService.Function2(result))
  ),

).subscribe([r1, r2, r3] => /* DO SOMETHING */);