roccomuso / batched-stream

Transform stream which batches a bunch of input data into groups of specified size (supporting objectMode).

Batched Stream

Transform stream supporting objectMode, which batches a bunch of input data into groups of specified size and will emit arrays, so that you can deal with pieces of input asynchronously.

Install

npm install --save batched-stream

Usage

const BatchStream = require('batched-stream')
const batch = new BatchStream({
  size : 5, // Bytes or N. of objects (when objectMode is true)
  objectMode: true, // false by default
  strictMode: false // return the rest of the batch when a stream contains a number of items that is not a strict multiply of the batch size
})

stream
  .pipe(batch)
  .pipe(new ArrayStream()) // In objectMode: true, deals with array input from pipe.

You can .pipe other streams to it or .write them yourself (if you .write don't forget to .end). The options are passed down to the transform stream (you can increase the highWaterMark for example).

• If objectMode is enabled the emitted batches are Array of length equal to size. Otherwise Buffer with byte size-length are emitted.

• The strictMode: false allows to get the rest of the batch when a stream contains a number of items that is not a strict multiply of the batch size.

Test

npm test

Author

Rocco Musolino (@roccomuso)

License

MIT

batched-stream's Issues

Suggestion: provide a maxWait param for cases where flush isn't enough

So, I was using this code (and thank you for making it) to convert a read Kafka stream into batches (to reduce concurrent load on the downstream destination) -

The issue is that it seems like some ReadableStreams don't actually ever invoke flush, making the strictMode irrelevant
as a replacement, i added a "maxWait" parameter that records the timestamp of the last update, and upon receiving a put into the BatchStream, checks whether the required batch size has been reached OR the maxWait has elapsed since the last put.

This way, there is an occasional internal "flushing" of the stream without waiting for an externally invoked flush...

wdyt?