How to get similar behavior to bufferCount whilst emitting if there are less items than the buffer count

EDIT: today I learned that bufferTime has a maxBufferSize parameter, that will emit when the buffer reaches that size. Therefore, the original answer below isn't necessary, we can simply do this:

const stream$ = subject$.pipe(
  bufferTime(2000, null, 3), // <-- buffer emits @ 2000ms OR when 3 items collected
  filter(arr => !!arr.length)
);

StackBlitz

ORIGINAL:

It sounds like you want a combination of bufferCount and bufferTime. In other words: "release the buffer when it reaches size X or after Y time has passed".

We can use the race operator, along with those other two to create an observable that emits when the buffer reaches the desired size OR after the duration has passed. We'll also need a little help from take and repeat:

const chunk$ = subject$.pipe(bufferCount(3));

const partial$ = subject$.pipe(
  bufferTime(2000),
  filter(arr => !!arr.length) // don't emit empty array
);

const stream$ = race([chunk$, partial$]).pipe(
  take(1),
  repeat()
);

Here we define stream$ to be the first to emit between chunk$ and partial$. However, race will only use the first source that emits, so we use take(1) and repeat to sort of "reset the race".

Then you can do your work with concatMap like this:

stream$.pipe(
  concatMap(chunk => this.doWorkWithChunk(chunk))
);

Here's a working StackBlitz demo.

You may want to roll it into a custom operator, so you can simply do something like this:

const stream$ = subject$.pipe(
  bufferCountTime(5, 2000)
);

The definition of bufferCountTime() could look like this:

function bufferCountTime<T>(count: number, time: number) {
  
  return (source$: Observable<T>) => {
    const chunk$ = source$.pipe(bufferCount(count));
    const partial$ = source$.pipe(
      bufferTime(time),
      filter((arr: T[]) => !!arr.length)
    );

    return race([chunk$, partial$]).pipe(
      take(1),
      repeat()
    );
  }
}

Another StackBlitz sample.

TLDR;

A StackBlitz app with the solution can be found here.

Explanation

Here would be an approach:

const bufferLen = 3;
const count$ = subject.pipe(filter((_, idx) => (idx + 1) % bufferLen === 0));
const timeout$ = subject.pipe(
  filter((_, idx) => idx === 0),
  switchMapTo(timer(0))
);

subject
  .pipe(
    buffer(
      merge(count$, timeout$).pipe(
        take(1),
        repeat()
      )
    ),
    concatMap(buffer => forkJoin(buffer.map(doWork)))
  )
  .subscribe(/* console.warn */);

/* Output:
Processing 1
Processing 2
Processing 3
Processed 1
Processed 2
Processed 3
Processing 4
Processing 5
Processed 4
Processed 5
Processing 6 <- after the `setTimeout`'s timer expires
Processing 7
Processing 8
Processed 6
Processed 7
Processed 8
*/

The idea was to still use the bufferCount's behavior when items come in synchronously, but, at the same time, detect when fewer items than the chosen bufferLen are in the buffer. I thought that this detection could be done using a timer(0), because it internally schedules a macrotask, so it is ensured that items emitted synchronously will be considered first.

However, there is no operator that exactly combines the logic delineated above. But it's important to keep in mind that we certainly want a behavior similar to the one the buffer operator provides. As in, we will for sure have something like subject.pipe(buffer(...)).

Let's see how we can achieve something similar to what bufferTime does, but without using bufferTime:

const bufferLen = 3;
const count$ = subject.pipe(filter((_, idx) => (idx + 1) % bufferLen === 0));

Given the above snippet, using buffer(count$) and bufferTime(3), we should get the same behavior.

Let's move now onto the detection part:

const timeout$ = subject.pipe(
  filter((_, idx) => idx === 0),
  switchMapTo(timer(0))
);

What it essentially does is to start a timer after the subject has emitted its first item. This will make more sense when we have more context:

subject
  .pipe(
    buffer(
      merge(count$, timeout$).pipe(
        take(1),
        repeat()
      )
    ),
    concatMap(buffer => forkJoin(buffer.map(doWork)))
  )
  .subscribe(/* console.warn */);

By using merge(count$, timeout$), this is what we'd be saying: when the subject emits, start adding items to the buffer and, at the same time, start the timer. The timer is started too because it is used to determine if fewer items will be in the buffer.

Let's walk through the example provided in the StackBlitz app:

from([1, 2, 3, 4, 5])
  .pipe(tap(i => subject.next(i)))
  .subscribe();

// Then mimic some more items coming through a while later
setTimeout(() => {
  subject.next(6);
  subject.next(7);
  subject.next(8);
}, 10000);

When 1 is emitted, it will be added to the buffer and the timer will start. Then 2 and 3 arrive immediately, so the accumulated values will be emitted.
Because we're also using take(1) and repeat(), the process will restart. Now, when 4 is emitted, it will be added to the buffer and the timer will start again. 5 arrives immediately, but the number of the collected items until now is less than the given buffer length, meaning that until the 3rd value arrives, the timer will have time to finish. When the timer finishes, the [4,5] chunk will be emitted. What happens with [6, 7, 8] is the same as what happened with [1, 2, 3].

Since I noticed the use of forkJoin in your sample code, I can see you are sending a request to the server for each emission (I was originally under the impression that you were making only 1 call per batch with combined data).

In the case of sending one request per item the solution is much simpler!

There is no need to batch the emissions, you can simply use mergeMap and specify its concurrency parameter. This will limit the number of currently executing requests:

const stream$ = subject$.pipe(
  mergeMap(val => doWork(val), 3),  // 3 max concurrent requests
);

Here is a visual of what the output would look like when the subject rapidly emits:

Notice the work only starts for the first 3 items initially. Emissions after that are queued up and processed as the prior in flight items complete.

Here's a StackBlitz example of this behavior.