angular - rxjs:如何让 foreach 循环等待内部 observable
问题描述
我有一个 foreach 循环,我可能需要一个 http 请求来获取一些信息。我尝试在 foreach 循环中使用 forkjoin 来“让 foreach 循环等待 observables”(在这个例子中只有一个 observables,但实际上我会有更多......)。但是使用此代码,每个 foreach 都会继续运行,而无需等待可观察对象完成,而且我找不到任何解决方案来防止这种情况发生。
所以非常感谢您的帮助!!!
...
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 循环的完整工作解决方案。在管道和数组上的映射函数中使用映射运算符要容易得多。我了解到,将这种逻辑拆分为多个运算符更容易,而不是尝试在 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 上调用 subscribe,除非您是最终消费者并且准备好丢弃 observable(不需要对数据进行更多转换)。这通常仅在向用户显示或写入数据库时发生。
你的代码在做什么(你的各种调用返回什么)对我来说不是很清楚。所以这是我将如何做你看起来正在尝试的事情的近似值。
这些方面的东西应该起作用。
请注意,在我准备好将最终的愿望清单记录到控制台之前,我是如何从不订阅的?这是设计使然,因此信息按预期进行。
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
s 的代码。我不认为这真的是一个问题,你可以在 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
toforkJoin(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[]>[])
);
}
推荐阅读
- types - Self referencing record types in Standard ML
- c++ - 带有模块化库的 C++ cmake 项目结构
- javascript - 使用 HTML 标签和引号解析 JSON
- firebase - Firebase Cloud Functions“数据请求超过最大大小”错误
- reactjs - 打字稿,反应错误 类型中缺少属性“[Invalid_propTypes]”
- android - X时间后杀死android应用程序
- python - 打开随机文件时出现UnicodeDecodeError错误?
- ruby-on-rails - 从 Faker 播种存储在 S3 中的个人资料图片
- mysql - 如何将我的本地 mysql 数据库连接到 MacBook Pro 中的 Apache 超集?
- android - 带有backstack的android架构导航OnNavigatedListener