Rxjs - Collects all unprocessed values

I have a custom operator that does this. Heads up, there are a few bits that make this more complicated than you might first think. On top of that, this is a bit over-engineered but it sounds like you can use this with the defaults to get what you want.

Custom RxJS Operator: bufferedExhaustMap

function bufferedExhaustMap<T, R>(
  project: (v: T[]) => ObservableInput<R>,
  minBufferLength = 0,
  minBufferCount = 1,
  concurrent = 1
): OperatorFunction<T, R> {

  function diffCounter(){
    const incOrDec = new Subject<boolean>();
    return {
      onAvailable: incOrDec.pipe(
        scan((acc, curr) => (curr ? ++acc : --acc), 0),
        startWith(0),
        shareReplay(1),
        first(count => count < concurrent),
        mapTo(true)
      ),
      start: () => incOrDec.next(true),
      end: () => incOrDec.next(false)
    };
  }

  return source => defer(() => {
    const projectCount = diffCounter();

    const shared = source.pipe(share());

    const nextBufferTime = () => forkJoin([
      shared.pipe(take(minBufferCount), delay(0)),
      timer(minBufferLength),
      projectCount.onAvailable
    ]);

    return shared.pipe(
      bufferWhen(nextBufferTime),
      delayWhen(() => projectCount.onAvailable),
      tap(projectCount.start),
      map(project),
      mergeMap(projected => from(projected).pipe(
        finalize(projectCount.end))
      )
    );
  });
}

Usage

Note, I've removed some superfluous promise/RxJS mingling as it's a code smell.

this.tasks.pipe(
  mergeMap___AllUnprocessedUntilNow(async tasks => {
    await this.processTasks(tasks)
  }, 1),
)

becomes

this.tasks.pipe(
  mergeMap___AllUnprocessedUntilNow(
    tasks => this.processTasks(tasks)
    , 1
  )
)

(remember that all high-order RxJS operators will turn iterables and promises into observables for you. Otherwise you can use the from operator instead).

Since bufferedExhaustMap has a default concurrency of 1, you can write this with the new operator as follows:

this.tasks.pipe(
  bufferedExhaustMap(
    tasks => this.processTasks(tasks)
  )
)

You can accomplish this with a combination of buffer and concatMap, along with a little help from filter and tap.

The idea is to trigger the release of the buffer at the correct time. In your case there are two occasions to release:

• When the first emission is received
• After work has been completed

const release$ = new Subject<void>();
let releaseOnNextEmit = true;

const work$ = item$.pipe(
  tap(() => { 
    if(releaseOnNextEmit) setTimeout(() => release$.next(), 0);
  }),
  buffer(release$),
  tap(tasks => releaseOnNextEmit = tasks.length === 0),
  filter(tasks => tasks.length > 0),
  concatMap(tasks => processTasks(tasks)),
  tap(() => release$.next())
);

You can see we use a Subject to trigger the release of the buffer and a releaseOnNextEmit flag to indicate whether receiving an emission should trigger the buffer's release.

After the buffer emits, we set the flag accordingly:

• true when buffer is empty
• false when it is NOT empty

filter is used to prevent passing along empty arrays to concatMap (buffer will emit empty array if the buffer is empty when its trigger emits).

We use concatMap to execute the real work, then simply release the buffer afterwards.

Here's a working StackBlitz demo.

With all the taps in there, this isn't super easy to follow, so it may be beneficial to put it into a custom operator:

function bufferConcatMap<T, R>(project: (a: T[]) => ObservableInput<R>) {
  const release$ = new Subject<void>();
  let releaseOnNextEmit = true;

  return (source$: Observable<T>) => source$.pipe(
    tap(() => { 
      if(releaseOnNextEmit) setTimeout(() => release$.next(), 0);
    }),
    buffer(release$),
    tap(tasks => releaseOnNextEmit = tasks.length === 0),
    filter(tasks => tasks.length > 0),
    concatMap(project),
    tap(() => release$.next())
  );
}

const work$ = item$.pipe(
  bufferConcatMap(tasks => processTasks(tasks))
);

Here's another StackBlitz.