Add stream block event

This is useful for example to access how many values were de/encoded.
mtth · May 2, 2021 · edb727e · edb727e
1 parent 69599b1
commit edb727e
Show file tree

Hide file tree

Showing 4 changed files with 2,736 additions and 180 deletions.
diff --git a/lib/containers.js b/lib/containers.js
@@ -243,7 +243,7 @@ BlockDecoder.prototype._writeChunk = function (chunk, encoding, cb) {
     nBlocks++;
     this._decompress(
       block.data,
-      this._createBlockCallback(block.count, chunkCb)
+      this._createBlockCallback(block.data.length, block.count, chunkCb)
     );
   }
   chunkCb();
@@ -255,7 +255,7 @@ BlockDecoder.prototype._writeChunk = function (chunk, encoding, cb) {
   }
 };
 
-BlockDecoder.prototype._createBlockCallback = function (count, cb) {
+BlockDecoder.prototype._createBlockCallback = function (size, count, cb) {
   var self = this;
   var index = this._index++;
 
@@ -266,6 +266,7 @@ BlockDecoder.prototype._createBlockCallback = function (count, cb) {
       self.emit('error', err);
       cb();
     } else {
+      self.emit('block', new BlockInfo(count, data.length, size));
       self._queue.push(new BlockData(index, data, cb, count));
       if (self._needPush) {
         self._read();
@@ -525,7 +526,7 @@ BlockEncoder.prototype._write = function (val, encoding, cb) {
 BlockEncoder.prototype._flushChunk = function (pos, cb) {
   var tap = this._tap;
   pos = pos || tap.pos;
-  this._compress(tap.buf.slice(0, pos), this._createBlockCallback(cb));
+  this._compress(tap.buf.slice(0, pos), this._createBlockCallback(pos, cb));
   this._blockCount = 0;
 };
 
@@ -551,7 +552,7 @@ BlockEncoder.prototype._read = function () {
   }
 };
 
-BlockEncoder.prototype._createBlockCallback = function (cb) {
+BlockEncoder.prototype._createBlockCallback = function (size, cb) {
   var self = this;
   var index = this._index++;
   var count = this._blockCount;
@@ -565,6 +566,7 @@ BlockEncoder.prototype._createBlockCallback = function (cb) {
       return;
     }
     self._pending--;
+    self.emit('block', new BlockInfo(count, size, data.length));
     self._queue.push(new BlockData(index, data, cb, count));
     if (self._needPush) {
       self._needPush = false;
@@ -576,6 +578,13 @@ BlockEncoder.prototype._createBlockCallback = function (cb) {
 
 // Helpers.
 
+/** Summary information about a block. */
+function BlockInfo(count, raw, compressed) {
+  this.valueCount = count;
+  this.rawDataLength = raw;
+  this.compressedDataLength = compressed;
+}
+
 /**
  * An indexed block.
  *