rxjs - 使用 combineLatest 将 n+1 个流添加到 RXJS 流

Question

希望有人能帮我解决这个问题。
我有 2 个流，需要在其上使用操作符 combineLatest。一段时间后，我需要添加也需要使用 combineLatest 的动态流。
这是我需要做的:

stream a ---------a---------------b-------------c------------------------>
stream b ----1--------2-----3-------------------------4------------------>
stream c (not defined at start)      -----z-----------------x------------>
stream d (not defined at start)                         ----------k------>
                                 (combineLatest)
result   ---------(a1)(a2)--(a3)--(b3)----(b3z)-(c3z)-(c4z)-(c4x)-(c4xk)->   

更新

更具体地说，我想转这个 STREAM (link)


对于这个结果:

A----B---B0-C0--D0--D1--E1--E1a--F1a--F2a---G2a---G3a--H3a-H3b--I3b

Answer 1

想法吧everything is a stream .甚至流:)

const onNew$ = new Rx.Subject();

const a$ = Rx.Observable.interval(1000).mapTo('a');
const b$ = Rx.Observable.interval(1000).mapTo('b');

const comb$ = Rx.Observable
  .merge(
    onNew$,
    Rx.Observable.from([a$, b$]),
  )
  .scan((acc, v) => {
      acc.push(v);
      return acc;
    }, [])
  .switchMap(vs => Rx.Observable.combineLatest(vs))

comb$.take(4).subscribe(v => console.log(v));

setTimeout(
  () => onNew$.next(Rx.Observable.interval(1000).mapTo('c')),
  2000,
);
setTimeout(
  () => onNew$.next(Rx.Observable.interval(1000).mapTo('d')),
  4000,
);

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>