var batch = new BatchStream({
  size : 100,     // the size for each chunk
  timeout: 5000   // emit data after this amount of milliseconds
                  // even if the size of buffered writes not reaching `size`
});
 
stream
  .pipe(batch)
  .pipe(new ArrayStream()); // deals with array input from pipe.
 

docs.on('data', function(doc) {
  db.insert(doc)
})

var batch = new BatchStream({
  transform: function(items, callback) {
    db.bulkInsert(items, callback)
  })
})
 
docs.pipe(batch)
.on('finish', function() {
  console.log('All doc inserted.')
})

var batch = new BatchStream()
 
docs.pipe(batch)
.on('data', function(items) {
  db.bulkInsert(items, ...)
})
.on('finish', function() {
  console.log('All docs queued for insertion.')
})