Angular 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

更新了添加的评论

另一个：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 2

我为你做了一个例子

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 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

Observable本质上是异步的。订阅调用将在另一个线程中运行，而方法中的其余指令将继续在主线程中运行。您需要做的是等待异步操作完成后再运行下一个。您可以通过在订阅中放置isntructions来实现这一点，或者更好地使用（错误）参数之后的onCompleted参数，就像这样

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