Would a bufferWhile operator have its place in the operator's list?

[LcsGa]

I had a use case today where I wanted to buffer some values based on a predicate!
Since I couldn't find an operator (maybe I missed it?) that already handled that I made one:

function bufferWhile<T, R extends T>(predicate: (value: T, index: number) => value is R): OperatorFunction<T, R[]>;
function bufferWhile<T>(predicate: (value: T, index: number) => boolean): OperatorFunction<T, T[]>;

function bufferWhile<T>(predicate: (value: T, index: number) => boolean): OperatorFunction<T, T[]> {
  return (source$: Observable<T>) =>
    new Observable((destination) => {
      let buffer: T[] = [];
      let index = 0;

      const emitBuffer = (firstValue?: T) => {
        destination.next(buffer);
        buffer = firstValue ? [firstValue] : [];
      };

      return source$.subscribe({
        next: (value) => {
          if (predicate(value, index++)) {
            buffer.push(value);
          } else {
            emitBuffer(value);
          }
        },
        error: (err) => destination.error(err),
        complete: () => {
          emitBuffer();
          destination.complete();
        },
      });
    });
}

Example:

of(1, 2, 3, 4, 5, 6)
  .pipe(bufferWhile((nb) => nb !== 4))
  .subscribe(console.log);
  // Outputs:
  // [1, 2, 3]
  // [4, 5, 6]

=> Does it have its place within the operators list?

BTW, did I miss some possible use cases with potential errors? I made this operator quickly and couldn't experiment more about it 😬

[LcsGa]

I improved it (I guess) by adding the same inclusive option as the takeWhile operator:

export function bufferWhile<T, R extends T>(predicate: (value: T, index: number) => value is R): OperatorFunction<T, R[]>;
export function bufferWhile<T, R extends T>(predicate: (value: T, index: number) => value is R, inclusive: boolean): OperatorFunction<T, R[]>;
export function bufferWhile<T>(predicate: (value: T, index: number) => boolean): OperatorFunction<T, T[]>;
export function bufferWhile<T>(predicate: (value: T, index: number) => boolean, inclusive: boolean): OperatorFunction<T, T[]>;

export function bufferWhile<T>(predicate: (value: T, index: number) => boolean, inclusive = false): OperatorFunction<T, T[]> {
  return (source$: Observable<T>) =>
    new Observable((destination) => {
      let buffer: T[] = [];
      let index = 0;

      const emitBuffer = (firstValue?: T) => {
        destination.next(buffer);
        buffer = firstValue ? [firstValue] : [];
      };

      return source$.subscribe({
        next: (value) => {
          if (predicate(value, index++)) {
            buffer.push(value);
          } else {
            if (inclusive) {
              buffer.push(value);
              emitBuffer();
            } else {
              emitBuffer(value);
            }
          }
        },
        error: (err) => destination.error(err),
        complete: () => {
          emitBuffer();
          destination.complete();
        },
      });
    });
}

[LcsGa]

any thought on this? @benlesh

[kwonoj]

We do carefully expand core api surface once if operator is suffeciently, widely used as well as does require some of tight integration to the core. You can publish operator as separate userland package to gauge usage & interest, but we won't add something into core right away.

[LcsGa]

Okay, I understand!
I already made a third-party lib including this one and others :)
I proposed this one though, simply because I felt that it was missing among the bufferXx family (thus to my feelings, this operator is legitimate to be present among them) => By that I mean that, even if that is not widely used, I thought that it may change if people had this new possibility they didn't have already (I may be entirely wrong though, plus I have a complete confidence the choices you make !)