如何正确地平坦这个可观察数组?

Question

我需要在Order[]中显示Observable<Order[]>[]的结果。

当前的操作方式:

const ordersObservable: Observable<Order[]>[] = [];
//ordersObservable is populated with a bunch of Observable<Order[]>
forkJoin(ordersObservable)
.pipe(
map((results) => ([] as Order[]).concat(...results))
)
.subscribe((orders: Order[]) => {
this.orderService.set(orders);
//...
});

我被告知我不应该像那样使用pipe，应该使用RxJs函数来处理它。

我试图使用concatAll和mergeAll而不是map，但这最终在ordersObservable中的每个项目调用orderService，而不是单个时间与ordersObservable的所有结果的平面数组。

我做错了什么?我如何在单个数组中平坦所有可观察的结果，最好是使用原生RxJs解决方案?

Answer 1

一个可观察对象是用来发出值的。对于一个Observable数组，我们需要订阅Observable数组中的每个元素来发出Order数组。

我真的认为你应该重新评估创建一个observable数组。如果你想对生成observable数组的代码提出另一个问题，我们可以提供建议。

也就是说，我能够让一些东西工作。我没有花时间去看看是否有更简单的方法。

from(this.ordersObservable)
.pipe(
concatAll(),
scan((acc, value) => [...acc, ...value], [] as Order[]),
takeLast(1)
)
.subscribe((x) => console.log('result', JSON.stringify(x)));

首先，我需要订阅一些东西。我们不能订阅observable数组。所以我用from把Observable的数组变成另一个Observable。这样我就可以订阅并执行代码。

from从数组中发出每个Observable。

注意:forkJoin和combineLatest也可以代替from，并且提供相同的结果。

然后使用concatAll()按顺序连接内部可观察对象。它订阅了数组中的每个Observable。

scan允许定义一个累加器，将每个order数组累加成一个order数组。

更新:添加takeLast(1)以确保只发出最后一个结果。

UPDATE 2: Second option

这也可以，并且可能更简单:

concat(...this.ordersObservable)
.pipe(
scan((acc, value) => [...acc, ...value], [] as Order[]),
takeLast(1)
)
.subscribe((x) => console.log('result', JSON.stringify(x)));

它使用数组解构("…")和concat运算符来发出observable数组中所有observable的所有值。然后我们仍然使用scan来累积它们，而takeLast(1)只排放一次。

我很想知道是否有人能够简化这个!