如何连接嵌套在另一个 Observable 中的多个 Observable 集合

Question

我是 RxJs 的新手，我正在为一个复杂的 Angular 应用程序而苦苦挣扎，在该应用程序中，我将 observables 嵌套在其他 observables 中（下文简化）。

简而言之，我有：

• 一个Observable，它返回一个带有对象集合的流SearchActivity => Observable<SearchActivity[]>
• 每个 SearchActivity 对象都有一个 Observable 属性 'results$' => BehaviorSubject<SearchActivityResult[]>

我尝试将所有活动中的所有结果提取到一个 Observable 中。 所以，从下面的例子来看（这不是代码，而是数据结构的一个想法）：

Observable<SearchActivity[]> => [
 {id: 1, results$: Observable<SearchActivityResult[]> => [1, 2]},
 {id: 2, results$: Observable<SearchActivityResult[]> => [3, 4, 5]} 
 {id: 3, results$: Observable<SearchActivityResult[]> => [6]} 
]

我想提取结果得到：

Observable<SearchActivityResult[]> => [1, 2, 3, 4, 5, 6] (Edit: the order is not important)

如何使用 RxJs 实现这一点？

到目前为止，我得到了类似的东西：

// This is not working properly
this.allResults$ = this.displayedActivities$.pipe(
  mergeMap(activities => {
    return concat(...activities?.map(activity => activity.results$ || of([])));
  })
);

但它似乎没有按预期工作。

我准备了StackBiltz with a more complex scenario here

2021-02-23 更新

我创建了一个新的StackBlitz，您可以通过单击它与结果进行交互。目标是获取每个数据库的所有结果的摘要（页面底部的表格），并显示所有搜索条件下正面/负面结果的数量。

TL;DR => 解决方案

我刚刚创建了一个复杂性较低的新StackBlitz，以隔离问题。我已经包含了Mrk Sef 提出的建议并且它正在工作。 我在平面运算符和对 json 的奇怪循环引用方面还有一些其他错误，但这更有可能是由 StackBlitz 模拟器引起的。

Answer 1

我认为您的想法是正确的，但您需要最后一步来将您的信息流还原为适合您的价值。

你写的：

this.allResults$ = this.displayedActivities$.pipe(
  mergeMap(activities => 
    concat(...activities?.map(
      activity => activity?.results$ || of([])
    ))
  )
);

activities 中的每个条目都有一个发射，但您不想要一系列发射，您想要一个合并/展平结果的数组。

你可以试试这样的：

this.allResults$ = this.displayedActivities$.pipe(
  mergeMap(activities => 
    merge(...activities?.map(
      activity => activity?.results$ || of([])
    ))
  ),
  toArray(),
  map(res => res.flat())
);

或

this.allResults$ = this.displayedActivities$.pipe(
  mergeMap(activities => 
    merge(...activities?.map(
      activity => activity?.results$ || of([])
    ))
  ),
  reduce((acc, curr) => [...acc,...curr], [])
);

---

更新：

你可以运行这个最小的例子：

const displayedActivities$ = of([
  {results$: of([1,2,3])},
  {results$: of([10,20,30])},
  {something: "else"},
  {results$: of([100,200,300])}
]);

const allResults$ = displayedActivities$.pipe(
  mergeMap(activities => 
    merge(...activities.filter(
      activity => activity?.results$ != null
    ).map(
      activity => activity.results$
    ))
  ),
  reduce((acc, curr) => [...acc,...curr], [])
).subscribe(console.log);

输出：

[1, 2, 3, 10, 20, 30, 100, 200, 300]

更新 #2

这就是您可以处理将属性 result$ observables 更改为长寿命的 observables 的方式。和上面的toArray()、Array.flat()差不多，只是现在用combineLatest

这是应该运行的代码。它将保留每个 observable 的最新发射作为最终组合输出的一部分：

模拟设置

创建displayedActivities$，它发出一个对象数组，每个对象都有一个属性result$，在完成之前发出一定次数（在本例中为3）。

/****
 * Pipeable Operator:
 * Takes arrays emitted by the source and spaces out their
 * values by the given interval time in milliseconds
 ****/
function intervalArray<T>(intervalTime = 1000): OperatorFunction<T[], T> {
  return pipe(
    concatMap((v: T[]) =>
      concat(
        ...v.map((value: T) =>
          EMPTY.pipe(
            delay(intervalTime),
            startWith(value)
          )
        )
      )
    )
  );
}

const displayedActivities$ = of([
  { results$: 
    of([
      [1,2,3],
      [4,5,6],
      [7,8,9]
    ]).pipe(intervalArray(250))
  },
  { results$: 
    of([
      [10,20,30],
      [40,50,60],
      [70,80,90]
    ]).pipe(intervalArray(300))
  },
  { something: "else" },
  { results$: 
    of([
      [100,200,300],
      [400,500,600],
      [700,800,900]
    ]).pipe(intervalArray(350))
  }
]);

合并所有结果$

const allResults$ = displayedActivities$.pipe(
  mergeMap(activities => 
    combineLatest(...activities.filter(
      activity => activity?.results$ != null
    ).map(
      activity => activity.results$.pipe(startWith([]))
    ))
  ),
  map(latest => latest.flat())
).subscribe(console.log);

输出：

[1,2,3]
[1,2,3,10,20,30]
[1,2,3,10,20,30,100,200,300]
[4,5,6,10,20,30,100,200,300]
[4,5,6,40,50,60,100,200,300]
[4,5,6,40,50,60,400,500,600]
[7,8,9,40,50,60,400,500,600]
[7,8,9,70,80,90,400,500,600]
[7,8,9,70,80,90,700,800,900]

Answer 2

您可以尝试使用 Array#map 映射数组中的 observables 并将其传递给 forkJoin 以进行并行请求。可以使用像 switchMap 这样的高阶映射运算符从源 observable 映射。

试试下面的

Observable<SearchActivity[]>.pipe(
  switchMap((data: any) => 
    forkJoin(data.map(item => 
      item.results$
    ))
  )
).subscribe({
  next: value => {
    console.log(value);
  },
  error: error => {
    // handle error
  }
});

注意：

1. Observable<SearchActivity[]> 这是一个占位符。将其替换为实际的源可观察变量。
2. forkJoin 仅在所有源 observables 完成时才会发出。如果您希望数据流替换为combineLatest 或zip。

Answer 3

这样的事情可能对你有用：

this.allResults$ = this.displayedActivities$.pipe(
  switchMap(activities => 
    merge(...activities.map(a => a.results$))
  ),                                         
  mergeMap(resultsArray => from(resultsArray)),
  scan((acc, cur) => ([...acc, cur]), [])
);

流程如下：

• 使用 merge 创建一个可在任何 activities.results$ 发射时发射的可观察对象
• 使用mergeMap / from() 分别发出数组元素
• 使用scan 将所有发射放入一个数组中

scan 内部的函数可能过于简单，因为它只是将所有发射组合到一个数组中，而没有考虑表示相同结果的多个发射。我看到你在返回的SearchActivityResult 项目上有一个id，所以也许可以在 scan 内部构建一个更复杂的处理函数来考虑这一点。

但希望这能让你开始。

这是一个有效的StackBlitz