RxJS 操作符 waitUntil

Question

a: 1---2-3-4--5---6
b: ------T---------

o: ------1234-5---6

使用RxJS，有没有什么算子可以完成上图？我有一个流 A，它是一个随机的事件流，给定一个流 B，它有一个 true事件，我可以有一个输出流，直到 true 才发出任何东西吗？事件，然后发送所有内容都已保存，然后再正常发出？

我想也许我可以用 buffer() ，但似乎没有办法用该运算符做这样的一次性缓冲区。

Answer 1

const { concat, interval, of, from } = rxjs;
const { share, delay, toArray, takeUntil, mergeMap } = rxjs.operators;

const waitUntil = signal$ => source$ => {
  const sharedSource$ = source$.pipe(share());
  return concat(
    sharedSource$.pipe(
      takeUntil(signal$),
      toArray(),
      mergeMap(from)
    ),
    sharedSource$
  );
}

const stopWaiting$ = of('signal').pipe(delay(2000));

const source$ = interval(500).pipe(
    waitUntil(stopWaiting$)
).subscribe(console.log);

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