Rxjs - 如何在数组中提取多个值并同步将它们反馈给可观察流

Question

我Rx.Observable从一系列事件中创建了一个:

Rx.Observable.fromEvent(recognizeStream, 'data')

其中每个数据事件如下所示:

{ error: null, alternatives: [result1, result2, result3] }

我想在数组中获取每个值alternatives并将它们合并到流中.我需要看哪些操作员？

据我所知,flatMap并且concatMap可以完成这项工作,但我从他们的例子中得不到这个想法.

有人可以解释我应该使用哪个运算符并向我提供一个示例吗？

Answer 1

xxxMap()运营商家族都处理高阶Observables.这意味着它们允许您在主Observable中创建Observable并将结果值内联到主流中.所以你可以读取类型签名Observable<Observable<T>> => Observable<T>

给定一个流,每个排放x是一个包含4个值排放的Observable:

input:  --x----------x 
flatMap   a-a-a-a-|  b-b-b-b-|
result: --a-a-a-a----b-b-b-b-|

类型转换xxxMap(myFnc)返回值

该xxxMap()运营商所有类型的工作成果Observable,Promise或Array.根据您输入的内容,如果需要,它将转换为Observable.

Rx.Observable.of('')
  .flatMap(() => [1,2,3,4])
  .subscribe(val => console.log('array value: ' + val));

Rx.Observable.of('')
  .flatMap(() => Promise.resolve(1))
  .subscribe(val => console.log('promise value: ' + val));

Rx.Observable.of('')
  .flatMap(() => Promise.resolve([1,2,3,4]))
  .subscribe(val => console.log('promise array value: ' + val));

Rx.Observable.of('')
  .flatMap(() => Rx.Observable.from([1,2,3,4]))
  .subscribe(val => console.log('Observable value: ' + val));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.3/Rx.js"></script>

在您的情况下,您可以轻松地flatMap对象并返回数组:

Rx.Observable.of({ error: null, alternatives: ['result1', 'result2', 'result3'] })
  .flatMap(val => val.alternatives)
  .subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.3/Rx.js"></script>

所有xxxMap运算符之间的差异

mergeMap做了什么

flatMap如,更好地mergeMap将merge所有排放每当他们进入主流.

concatMap

concatMap将concat在下一个流之前等待所有排放完成:

switchMap

但是当新的发射可用时,switchMap将放弃上一个流,并切换为从新流发出值: