Angular/RxJS中不使用toArray()获取Observable末尾数组的方法

你可以使用RxJS的reduce操作符实现相同效果，它会在源Observable完成时，将所有发射项累积为一个数组，行为和toArray()完全一致。

修改后的代码 #

private enriched_sns$ = toObservable(this.sn_results).pipe(
  concatMap((results) => results as StreamNotificationResult[]),
  concatMap((notification) => {
    const user_id = notification.activities[0].actor.id;
    return zip(of(notification), this.read_user_data_service.readUserData(user_id));
  }),
  map(([notification, user]) => ({
    notification,
    user
  })),
  // 用reduce替代toArray()，累积所有项为数组
  reduce((accumulator, currentItem) => {
    accumulator.push(currentItem);
    return accumulator;
  }, [] as { notification: StreamNotificationResult; user: any }[])
);

原理说明 #

• reduce接收两个参数：累加器函数和初始值（这里是空数组）。
• 每次上游发射新的处理后对象时，累加器函数会把当前项加入数组并返回更新后的数组。
• 当上游所有Observable都完成时，reduce会发出最终的完整数组，这和toArray()的输出完全匹配。

另一种可选方案（scan + takeLast） #

如果你需要在过程中也能看到累积的中间数组，可以用scan结合takeLast(1)，但这种方式会多次发射中间结果，最后只取完整数组：

private enriched_sns$ = toObservable(this.sn_results).pipe(
  concatMap((results) => results as StreamNotificationResult[]),
  concatMap((notification) => {
    const user_id = notification.activities[0].actor.id;
    return zip(of(notification), this.read_user_data_service.readUserData(user_id));
  }),
  map(([notification, user]) => ({
    notification,
    user
  })),
  // 每次发射累积的数组
  scan((accumulator, currentItem) => [...accumulator, currentItem], [] as { notification: StreamNotificationResult; user: any }[]),
  // 只取最后一次的完整数组
  takeLast(1)
);

注意：这两种方案都依赖上游Observable最终会完成，和toArray()的要求一致——如果上游不完成，就不会发出结果。

内容的提问来源于stack exchange，提问作者syahiruddin