chunking_object_stream.js

class ChunkingObjectStream extends WriteableStream {
  
  constructor(options) {
    options.objectMode = true;
    super(options);
    this.chunkSize = options.chunkSize || 100;
    this.queue = [];
    this.counter = 0;
    this.emitted = 0;

    this.once('finish', () => {
      this._flushOnFinish();
    });
  }

  _write(chunk, encoding, callback) {
    this.counter++;

    if(this.queue.length == this.chunkSize) {
      this._flushQueue();
    }

    this.queue.push(chunk);
    callback();
  }

  _flushQueue() {
    this.emitted += this.queue.length;
    this.emit('chunk', this.queue, this.emitted);
    this.queue = [];
  }

  _flushOnFinish() {
    this._flushQueue();
    this.emit('done', this.counter, this.emitted);
  }
}