Rxjs: как заставить цикл foreach ждать внутреннего наблюдаемого

Для этого вы используете from и concatMap.

let wishes: Wish[] = [];
from(response.wishes).pipe(concatMap(wish) => {
      let newWish : Wish = new Wish( wish._id, wish.title, wish.price );
               
      let personObservables: Observable<any>[] = [];
      if (wish.reservation){
        personObservables.push(this.peopleService.getPersonById(wish.reservation.reservedBy)); 
      } else {
        personObservables.push(of(null));
      }
      return forkJoin(personObservables).pipe(tap([reservedBy]) => {
        if (reservedBy) {
          newWish.reservation = {
            ...wish.reservation,
            reservedBy
          };
        }
        wishes.push(newWish);
      }));
    }).subscribe();

вы можете создать эту функцию для обработки вашего процесса:

completeData(items): Observable<any> {
    let $apiCallItems: Observable<any>[] = [];
    const itemsNeedServiceCall: Array<any> = [];
    const itemsCompleted: Array<any> = [];
    items.forEach((item) => {
      if (item.reservation) {
        $apiCallItems.push(this.peopleService.getPersonById(item.reservation.reservedBy));
        itemsNeedServiceCall.push(item);
      } else {
        itemsCompleted.push(item);
      }
    });
    return forkJoin($apiCallItems).pipe(
      map((r) => {
        for (let i = 0; i < r.length; i++) {
          itemsNeedServiceCall[i] = {
            ...itemsNeedServiceCall[i],
            ...r[i],
          };
        }
        return [...itemsCompleted, ...itemsNeedServiceCall];
      })
    );
  }

а затем в любом месте, которое вы хотите использовать, вы можете сделать следующее:

this.completeData(response.wishes).subscribe(r => {
    //some code
})

Как правило, никогда не вызывайте подписку на наблюдаемое, если только вы не являетесь конечным потребителем и не готовы выбросить наблюдаемое (больше никаких преобразований в данные не требуется). Обычно это происходит только при отображении пользователю или записи в базу данных.

Что делает ваш код (что возвращают ваши различные вызовы), мне не очень понятно. Итак, это приблизительное представление о том, как я буду делать то, что вы пытаетесь сделать.

Что-то в этом духе должно работать.

Заметили, что я никогда не подписываюсь, пока не буду готов записать окончательный список пожеланий в консоль? Так задумано, поэтому информация соответствует ожидаемому.

return this.http$.get<IWishlistResponse[]>(
  environment.apiUrl + 
  'wishlist/' + 
  receiver.id
).pipe(
  map(wishlist => wishlist.wishes.map(wish => ({
    oldWish: wish,
    newWish: new Wish( wish._id, wish.title, wish.price)
  }))),
  map(wishes => wishes.map(({oldWish, newWish}) => 
    !oldWish.reservation ? 
    of(newWish) :
    this.peopleService.getPersonById(oldWish.reservation.reservedBy).pipe(
      map(([reservedBy]) => reservedBy ?
        {
          ...newWish, 
          reservation: {
            ...oldWish.reservation,
            reservedBy
          }
        } :
        {
          ...newWish,
          reservation: oldWish.reservation
        }
      )
    )
  )),
  switchMap(newWishObservables => forkJoin(newWishObservables))
).subscribe(newWishes => console.info(newWishes));

Обновление №1: Очистка кода

Некоторые вещи, которые упрощают обслуживание кода, проверку ошибок и т. д., заключаются в том, чтобы передать создание наблюдаемых объектов в отдельные функции и сохранить конвейеры «чистыми (er)» от логики преобразования.

Как только вы это сделаете, этот шаблон:

map(things => things.map(thing => observableThing)),
mergeMap(observableThings => forkJoin(observableThings))

можно комбинировать, не создавая лишнего беспорядка

mergeMap(things => forkJoin(
  things.map(thing => observableThing)
))

Вот как я могу навести порядок в вашем обновленном коде (это делается быстро на лету, не тестировалось).

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

Вы также заметите, что я быстро обработал ошибки в обеих функциях, которые создают наблюдаемые объекты. Оба раза я просто заменял значение по умолчанию (хотя и в разных частях трубы с немного другим эффектом). На практике вы, вероятно, захотите проверить имя или тип ошибки и отреагировать соответствующим образом.

Я просто игнорирую ошибки, но некоторые ошибки нельзя игнорировать спокойно. Это зависит от вас.

public getWishlist ( receiver : Person) : Observable<Wish[]> {
  return this.http$.get<IWishlistResponse[]>(
    environment.apiUrl + 'wishlist/' + receiver.id
  ).pipe(

    // Create wish instances from each wish in API response and save reservation for later use
    map( wishlistResponse => 
      wishlistResponse[0].wishes.map(wish => ({
        wish: this.createWishInstanceFromResponse(wish, receiver),
        reservation: wish.reservation
      }))
    ),
    
    // For each wish with reservation: get person info for 'reservedBy' id
    map(wishesAndReservationObjects => 
      wishesAndReservationObjects.map(({wish, reservation}) => 
        this.addReservationToWish(wish, reservation)
      )
    ),
    
    // forkJoin all observables, so the result is an array of all the wishes
    switchMap(reservedByObservables => 
      forkJoin(reservedByObservables)
    ),
    
    // Call method on each wish (with or without reservation) to set user flags in each instance (must be done after reservedBy is added)
    map (wishes => wishes.map(wish => {
      wish.setUserIsFlags(this.userService.currentUser);
      return wish;
    })),
    
    // For each wish: get state via API call
    map(wishesWithoutState => wishesWithoutState.map(wishWithoutState => 
      this.addStatusToWish(wishWithoutState)
    )),
    
    // Combine all stateObservables into 1 array
    switchMap(stateObservables =>
      forkJoin(stateObservables)
    ),

    // If this.http$.get<IWishlistResponse[]> is empty, this will all
    // complete without emitting anything. So we can just give it a default
    // thing to emit in case that happens.
    defaultIfEmpty(<Wish[]>[])
  );
}

private createWishInstanceFromResponse( 
  wishResponse : IWishResponse, 
  receiver: Person 
): Wish {
  let wish : Wish = new Wish (
    wishResponse._id,
    wishResponse.title,
    wishResponse.price
  );
  return wish;
}

/***
 * Create an observable that tries to add reservation and 
 * reservedBy to the wish
 ***/ 
private addReservationToWish(wish: Wish, reservation): Observable<Wish>{
  return this.peopleService.getPersonById(reservation?.reservedBy).pipe(
    // if peopleService.getPersonById fails (Say, because reservation was 
    // null), convert the error into a null response, the filter will Ignore
    // the failed call and just return wish.
    catchError(_ => of(null)),
    // Since errors become null, this filters errors. Furthermore if peopleService
    // doesn't error and just emits a null/undefined value instead, this will
    // filter that situation as well
    filter(reservedBy => reservedBy != null),
    // Now we know reservedBy isn't null, so map doesn't need to check
    map(reservedBy => { 
      wish.reservation = reservation;
      wish.reservation.reservedBy = new Person(
        reservedBy.id, 
        reservedBy.firstName, 
        reservedBy.lastName,
        ...
      );
      return wish;
    }),
    // If reservedBy was null (due to error or otherwise), then just emit
    // the wish without a reservation
    defaultIfEmpty(wish)
  );
}

/***
 * Create an observable that tries to add status to the wish
 ***/ 
private addStatusToWish(wish: Wish): Observable<Wish>{
  return this.http$.get<WishStatus>(
    environment.apiUrl + 
    'wish/' + 
    wishWithoutState.id + 
    '/state'
  ).pipe(
    map(state => ({
      ...wish,
      status: state
    })),
    // If http$.get<WishStatus> failed, then just return the wish
    // without a status
    catchError(_ => of(wish)),
  );
}

Теперь, когда переходы от map к forkJoin(mapped) выглядят достаточно чистыми, я бы, наверное, объединил их, но это на ваше усмотрение.

Вот как это будет выглядеть:

public getWishlist ( receiver : Person) : Observable<Wish[]> {
  return this.http$.get<IWishlistResponse[]>(
    environment.apiUrl + 'wishlist/' + receiver.id
  ).pipe(

    // Create wish instances from each wish in API response
    // For each wish with reservation: get person info for 'reservedBy' id
    switchMap(wishlistResponse => forkJoin(
      wishlistResponse[0].wishes.map(wish => 
        this.addReservationToWish(
          this.createWishInstanceFromResponse(wish, receiver), 
          wish.reservation
        )
      )
    )),
    
    // Call method on each wish (with or without reservation) to set user flags in each instance (must be done after reservedBy is added)
    map (wishes => wishes.map(wish => {
      wish.setUserIsFlags(this.userService.currentUser);
      return wish;
    })),
    
    // For each wish: get state via API call
    switchMap(wishesWithoutState => forkJoin(
      wishesWithoutState.map(wishWithoutState => this.addStatusToWish(wishWithoutState))
    )),

    // If this.http$.get<IWishlistResponse[]> is empty, this will all
    // complete without emitting anything. So we can just give it a default
    // thing to emit in case that happens.
    defaultIfEmpty(<Wish[]>[])
  );
}