javascript - RxJS promise 链

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*/);