问题描述
我有一个 foreach 循环，我可能需要一个 http 请求来获取一些信息。我尝试在 foreach 循环中使用 forkjoin 来“使 foreach 循环等待 observables”（在这个例子中只有一个 observable，但实际上我会有更多......）。但是使用此代码，foreach 每个都继续运行而无需等待 observables 完成，我找不到任何解决方案来防止这种情况发生。
非常感谢您的帮助！！！
...
let wishes: Wish[] = [];
response.wishes.forEach((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));
}
forkJoin(personObservables).subscribe( ([reservedBy]) => {
if (reservedBy) {
newWish.reservation = {
...wish.reservation,reservedBy
};
}
wishes.push(newWish);
} );
});
...
编辑：没有 foreach 循环的完整工作解决方案。在管道中使用 map 运算符和在数组上使用 map 函数要容易得多。我了解到将这种逻辑拆分到多个运算符中比尝试在 1 个映射运算符中全部修复更容易...
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}) =>
!reservation ?
of(wish) :
this.peopleService.getPersonById(reservation.reservedBy)
.pipe(
map ( reservedBy => {
if (reservedBy) wish.reservation = {
...reservation,reservedBy: new Person(reservedBy.id,reservedBy.firstName,reservedBy.lastName)
}
return wish;
})
)
)),// forkJoin all observables,so the result is an array of all the wishes
switchMap(reservedByObservables => reservedByObservables.length !== 0 ? forkJoin(reservedByObservables) : of(<Wish[]>[])),//https://stackoverflow.com/questions/41723541/rxjs-switchmap-not-emitting-value-if-the-input-observable-is-empty-array
// 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.http$.get<wishStatus>(environment.apiUrl + 'wish/' + wishWithoutState.id + '/state')
.pipe(
catchError(() => of(null)),map( state => {
wishWithoutState.status = state;
return wishWithoutState;
})
)
)),// Combine all stateObservables into 1 array
switchMap(stateObservables => stateObservables.length !== 0 ? forkJoin(stateObservables) : of(<Wish[]>[]))
)
}
private createWishInstanceFromresponse ( wishResponse : IWishResponse,receiver: Person ) : Wish {
let wish : Wish = new Wish (
wishResponse._id,wishResponse.title,wishResponse.price,...
);
return wish;
}
解决方法
作为一般规则，除非您是最终消费者并且您准备将 observable 丢弃（不再需要对数据进行转换），否则永远不要在 observable 上调用 subscribe。这通常仅在向用户显示或写入数据库时发生。
我不太清楚你的代码在做什么（你的各种调用返回什么）。所以这是我将如何做你似乎正在尝试的近似。
应该按照这些方法行事。
注意我是如何在准备好将最终愿望列表记录到控制台之前从不订阅的？这是设计使然，因此信息符合预期。
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.log(newWishes));
更新 # 1：清理代码
使代码更易于维护、错误检查等的一些事情是将可观察对象的创建外包到单独的函数中，并保持转换逻辑的管道“干净（更）”。
一旦你这样做了，这个模式：
map(things => things.map(thing => observableThing)),mergeMap(observableThings => forkJoin(observableThings))
可以组合在一起而不会造成太多混乱
mergeMap(things => forkJoin(
things.map(thing => observableThing)
))
以下是我在更新代码中整理管道的方法（这只是即时完成，未经过测试）。
您会注意到我删除了您处理空 forkJoin 的代码。我不认为这真的是一个问题，您可以稍后在 observable 完成时在管道中解决该问题。
您还会注意到，我在创建 observable 的两个函数中都做了一些快速的错误处理。两次我都只替换了一个默认值（尽管在管道的不同部分，效果略有不同）。在实践中，您可能需要检查错误的名称或类型并做出相应的响应。
我只是忽略了错误，但不应该悄悄忽略一些错误。这取决于你。
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,wish.reservation
)
)
)),// For each wish: get state via API call
switchMap(wishesWithoutState => forkJoin(
wishesWithoutState.map(wishWithoutState => this.addStatusToWish(wishWithoutState))
)),this will all
// complete without emitting anything. So we can just give it a default
// thing to emit in case that happens.
defaultIfEmpty(<Wish[]>[])
);
}
您使用 from 和 concatMap 来完成此操作。
let wishes: Wish[] = [];
from(response.wishes).pipe(concatMap(wish) => {
let newWish : Wish = new Wish( wish._id,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
})