Buffer Count Or Time operator

[nordfjord]

After submitting a few PR's I thought it might be a good idea to share an operator that's been tremendously useful for me.

The scenario I'm dealing with is a subscription (i.e. to a message broker) that emits pretty fast. I'm handling messages from this queue in batches.

Using a plain old bufferCount left me in trouble because messages slow down in the evening. During the slowdown some messages would wait a long time in the buffer before being handled (sometimes not until the next day 😱)

I needed an operator that would "flush" the buffer every so often in case of slow messages.

Introducing bufferCountOrTime 🎉

I changed my code from:

subscription.pipe(
  bufferCount(16)
  // ...

to

subscription.pipe(
  bufferCountOrTime(16, 1000)
  // ...

And all is well! At worst a message waits a second in the buffer before being handled 🚀

[nordfjord]

Let me know if this is something that would better belong in a contrib library and not here

[trxcllnt]

Sorry for the delay. This looks great, thanks @nordfjord!

[trxcllnt]

Released in ix@4.4.0.