rxjs mergeall

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

// 创建一个高阶可观察对象，它发出三个内部可观察对象
const higherOrderObservable = of(
  interval(1000).pipe(take(3)), // 发出 0, 1, 2
  interval(1500).pipe(take(2)), // 发出 0, 1
  interval(2000).pipe(take(1))  // 发出 0
);

// 使用 mergeAll 将高阶可观察对象转换为单一的可观察对象
higherOrderObservable.pipe(
  mergeAll()
).subscribe(value => console.log(value));

// 输出将是：0, 0, 0, 1, 1, 2（顺序可能因并发性而有所不同）

import { of, interval } from 'rxjs';
import { mergeAll, take, catchError } from 'rxjs/operators';

const higherOrderObservable = of(
  interval(1000).pipe(take(3)),
  interval(1500).pipe(take(2)),
  interval(2000).pipe(take(1), catchError(err => of('Error occurred')))
);

higherOrderObservable.pipe(
  mergeAll()
).subscribe({
  next: value => console.log(value),
  error: err => console.error(err)
});

// 即使某个内部可观察对象出错，流也不会中断