RxJS 承诺链式

Question

我对 RxJS 还是很陌生。所以我有一个为不同的 ajax 调用创建有效负载的流，然后我使用 flatMap 来检索我需要的数据，它工作正常。很简单。

const streamA = Rx.Observable.from(array);

const streamB = streamA
  .map( val => /* build payload */ )
  .flatMap( payload => Rx.Observable.fromPromise($.ajax(payload))

streamB.subscribe( result => /* got it */)

现在我将为每个项目创建一个有效负载数组，但问题是现在当我订阅流时，我会返回每个请求，但完成后我只会返回初始元素。

const streamC = streamA
  .flatMap( payloads => {
    return Rx.Observable.from(payloads)
      .flatMap( payload => Rx.Observable.fromPromise($.ajax(payload))

streamC.subscribe( result => /* executed for every payload */)

我尝试使用 groupBy 返回正确的分组数组并向我展示我可以链接 Observable，但我仍然无法弄清楚如何正确订阅观察者以实现元素。

const streamWLF = streamA
  .flatMap( payloads => {
    return Rx.Observable.from(payloads)
      .flatMap( payload => Rx.Observable.fromPromise($.ajax(payload))
      .groupBy((obs) => obs.key, (obs) => obs)

streamWLF.subscribe( result => {
  result.subscribe(/* did my magic here*/);
})

所以我的问题是，最好的方法是什么？

收到子流时是否总是触发订阅的主流？

如果可能的话，我如何订阅 subStream 以触发仅在 subStream 完成时订阅的 mainStream？

Answer 1

我认为您正在寻找 forkJoin 运算符。

const streamC = streamA
  .flatMap(payloads => Rx.Observable.forkJoin(payloads));

streamC.subscribe( results => /*An array containing the results from all payloads*/);