Add a bufferUntil operator

[Sawtaytoes]

Just this week, I had 3 completely different RxJS projects need a bufferUntil operator. Two of those projects are waiting on auth tokens and need to buffer actions until it's available. The other one needs to wait on some files to be cached before allowing code to run against those files.

bufferUntil would buffer requests until a notifier emits, then no longer buffers after that.

Since I've never written code in the RxJS library, here's my userland interpretation of how I'd build this operator:

// Renamed `window` to `segment`: https://github.com/ReactiveX/rxjs/discussions/6054
const bufferUntil = (
  notifier$,
) => (
  pipe(
    segment(
      notifier$
    ),
    mergeMap((
      segment$,
      index,
    ) => (
      index === 0
      ? (
        segment$
        .pipe(
          toArray(),
          mergeAll(),
        )
      )
      : (
        segment$
      )
    )),
  )
)

If I had to rewrite this to match other buffer methods, it'd look like this:

// ...
index === 0
? (
  segment$
  .pipe(
    toArray(),
  )
)
: (
  segment$
  .pipe(
    map((
      value,
    ) => [
      value,
    ])
  )
)
// ...

Many people use Redux-Observable, and waiting for an authToken or valid cookie before making API requests is a common pattern. I can't imagine I'm the only person who's ever run into this use case.

Questions

1. Do others find this valuable? (of course you do 😉)
2. Do we want mergeAll or is it possible we'd want this to be customizable so someone could use switchAll or exhaustAll? I can't think why you'd want those, but there's no way to first() or last() since you wouldn't have access to the segment.
3. With bufferUntil, you might expect to get an array back like the other bufferX methods. Doesn't it make more sense to get a stream of values like my first code example rather than an array?

[raymonddavis]

@Sawtaytoes Does something like this help? It would need some kind of max thought because it can not buffer values forever or the inner items array would become huge.
https://stackblitz.com/edit/rxjs-bufferuntil?file=index.ts

[voliva]

Inpired from @raymonddavis I made this version:

https://stackblitz.com/edit/rxjs-bufferuntil-73pua4?file=index.ts

I think it's a bit more versatile: Instead of having a stream that you listen just once, you pass in an observable of booleans, which tell if it should emit or keep buffering.

I'm not sure I'd vote to have this bundled in rxjs/operators though... I guess it could be useful, but it will help making rxjs' API even bigger.

[raymonddavis]

Inspired by @voliva
https://stackblitz.com/edit/rxjs-bufferuntil-wexdim?file=index.ts

I moved the emit around because I was worried about a new value coming in while the buffer was emitting its values.

I like the toggle support and I agree that this being bundled into rxjs/operators doesn't seem worth it. If this was long-running the array would only grow in size forever, without a buffer size limit, this could cause issues.