Угловые 7 вложенных наблюдаемых

Question

У меня есть две коллекции: люди и домашние животные.Каждый питомец имеет личность.Моя цель - собрать всех людей и добавить их / ее питомцев в один JSON.Что я сделал до сих пор:

this.personService.getPersions().subscribe(persons => {
  const personsWithPets = persons.flatMap(person => this.petService.getPetsByPersonId(person._id)
    .subscribe(petsData => {
      person.pets = petsData;
      return person;
    }, (err) => {
     console.log(err);
    }));
  this.persons = personsWithPets; // This is fired before previous subscribe
}, (err) => {
  console.log(err);
});

Что я делаю не так?Почему this.persons = personsWithPets; срабатывает до окончания подписки?

Answer 1

Я сделал для вас пример

const { of } = rxjs;
const { map, switchMap, toArray, mergeMap } = rxjs.operators;

function getPeople() {
  return of([{ id: 1, name: 'Amanda' }, { id: 2, name: 'Nancy' }]);
}

function getPetsByPersonId(id) {
  switch(id) {
    case 1:
      return of(['Doggie']);
    case 2:
      return of(['Kitten']);
  }
}

const getPets = (person) => {
  return getPetsByPersonId(person.id).pipe(
    map(pets => ({ ...person, pets }))
  )
}

getPeople().pipe(
  switchMap(people => people),
  mergeMap(getPets),
  toArray()
)
  .subscribe(peopleWithPets => console.log(peopleWithPets));

<script src="https://unpkg.com/rxjs/bundles/rxjs.umd.min.js"></script>

Answer 2

Обновлено добавлены комментарии

Еще один: stackblitz

this.service.getPersons().pipe(switchMap((per:any[])=>{
       //create an array of observables
       const obs=per.map(per=>this.service.getPet(per.id));
       //call all of them in forkjoin
       return forkJoin(obs).pipe(map(pets=>
         //pets is an array, in pets[0] is the response of getPet(1), 
         //in pets[1] is the response of getPet(2)
         pets.map((pet,i)=>{
           return {
             ...per[i], //all the properties of the person
             pets:pet   //+ in pets an array with the pets of the person
             }
         })
       ))
     })).subscribe(res=>this.res=res)

Answer 3

Использование подписки внутри подписки считается плохой практикой и может привести к проблемам, как вы описали. Я бы предложил использовать mergeMap в сочетании с оператором forkJoin:

this.personService.getPersions().pipe(mergeMap(persons => {
   const requests = persons.map(person => this.petService.getPetsByPersonId(person._id));
   return forkJoin(of(persons), ...requests);
}),
map(values => {
  const persons = values[0];
  const pets = values.slice(1);
  // here you need to assign correct pet to correct person

})
).subscribe(personsWithPets => {
  console.log(personsWithPets);
}, err => {
  console.log(err);
});

Answer 4

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

this.personService.getPersons().subscribe(persons => {
  const personsWithPets = persons.flatMap(person => 
        this.petService.getPetsByPersonId(person._id)
            .subscribe(petsData => {
                 person.pets = petsData;
                 return person;
             }, 
             (err) => {
                 console.log(err);
             },
             () => {
                 this.persons = personsWithPets;
             }))
        });