RxJS mergeAll

將高階 Observable 轉(zhuǎn)換為其中的一階 Observable 同時傳遞內(nèi)部 Observable 上發(fā)出的所有值。

mergeAll<T>(concurrent: number = Number.POSITIVE_INFINITY): OperatorFunction<ObservableInput<T>, T>

參量

returns

OperatorFunction<ObservableInput<T>, T>：發(fā)出所有值的 Observable 源 Observable 發(fā)出的內(nèi)部Observable。

描述

展平可觀察的事物。

mergeAll訂閱發(fā)出 Observable 的 Observable，也稱為 一個更高階的 Observable。 每次觀察到其中一種發(fā)射出的內(nèi)部 可觀察對象，它訂閱并從 內(nèi)部 Observable 在輸出 Observable上。 僅輸出可觀察 一旦所有內(nèi)部可觀測對象都完成，則完成。 傳遞的任何錯誤 內(nèi)部 Observable 將立即在輸出 Observable 上發(fā)出。

例子

為每個點(diǎn)擊事件產(chǎn)生一個新的 Observable 間隔，并將其輸出混合為一個 Observable

import { fromEvent, interval } from 'rxjs';
import { map, mergeAll } from 'rxjs/operators';


const clicks = fromEvent(document, 'click');
const higherOrder = clicks.pipe(map((ev) => interval(1000)));
const firstOrder = higherOrder.pipe(mergeAll());
firstOrder.subscribe(x => console.log(x));

每次點(diǎn)擊從每秒 0 到 9 進(jìn)行計數(shù)，但僅允許 2 個并發(fā)計時器

import { fromEvent, interval } from 'rxjs';
import { take, map, mergeAll } from 'rxjs/operators';


const clicks = fromEvent(document, 'click');
const higherOrder = clicks.pipe(
  map((ev) => interval(1000).pipe(take(10))),
);
const firstOrder = higherOrder.pipe(mergeAll(2));
firstOrder.subscribe(x => console.log(x));

也可以看看

• combineAll
• concatAll
• exhaust
• merge
• mergeMap
• mergeMapTo
• mergeScan
• switchAll
• switchMap
• zipAll