用于可观察链接的仓促 forkjoin 替代 rxjs？

Question

我有 5 个不同的 API 调用，它们现在都链接在 forkJoin 中。我的新要求是 subscribe 应该在任何新的 observable 解决时触发。

在 rxjs 中是否有任何运算符或任何其他技巧可以保持链接，但是每次任何可观察的解决方案都应该触发它？

forkJoin(
        this.myService.api1(),
        this.myService.api2(),
        this.myService.api3(),
        this.myService.api4(),
        this.myService.api5()
    )
        .subscribe(
            ([r1,r2,r3,r4,r5]) => { ... do something })

Answer 1

您可以merge像 forkJoin 一样同时执行您的可观察量，但立即发出它们的值。为了跟踪顺序，将可观察量的索引添加到其输出中map。用于scan跟踪先前的值，将当前值插入数组中的正确位置并发出累积的数据。

export function forkJoinEarly(...sources: Observable<any>[]): Observable<any[]> {
  return merge(...sources.map((obs, index) => obs.pipe(
    // optional: only emit last value like forkJoin
    last(), 
    // add the index of the observable to the output
    map(value => ({ index, value })) 
  ))).pipe(
    // use scan to keep track of previous values and insert current values
    scan((acc, curr) => (acc[curr.index] = curr.value, acc), Array(sources.length).fill(undefined))
  );
}

https://stackblitz.com/edit/rxjs-gwch8m