RxJS оператор waitUntil

Question

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

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

Используя RxJS, есть ли оператор, который может выполнить диаграмму выше? У меня есть поток A, который является случайным потоком событий, учитывая поток B, который имеет единственное событие true, могу ли я иметь выходной поток, который ничего не излучает до этого события true, а затем отправляет все, что было сохраняется до тех пор, а потом излучает нормально?

Я подумал, что, возможно, я мог бы использовать buffer(), но, похоже, нет способа создать одноразовый буфер, подобный этому, с этим оператором.

Answer 1

Я думаю, что решение @ ZahiC правильное, но лично я бы сделал это в одной цепочке, используя оператор multicast.

a$.pipe(
  multicast(new Subject(), s => concat(
    s.pipe(
      buffer(b$),
      take(1),
    ),
    s
  )),
)

multicast будет в основном разделять поток на два, где concat будет сначала подписываться на первый, который буферизуется, пока не выдаст b$. Затем он завершается немедленно, потому что take(1) и concat снова подписываются на тот же поток, но на этот раз без буферизации.

Answer 2

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>