feat: add async chunk compress (#6)

Author: g11tech

index.js

@@ -5,7 +5,7 @@ module.exports = {
     createUncompressStream: function (opts) {
       return new UncompressStream(opts)
     }
-  , createCompressStream: function () {
-      return new CompressStream()
+  , createCompressStream: function (opts) {
+      return new CompressStream(opts)
     }
 }

lib/compress-stream.js

@@ -20,10 +20,11 @@ var Transform = require('stream').Transform
   , COMPRESSED = bufferFrom([ 0x00 ])
   , UNCOMPRESSED = bufferFrom([ 0x01 ])
 
-  , CompressStream = function () {
+  , CompressStream = function (opts) {
       if (!(this instanceof CompressStream))
-        return new CompressStream()
+        return new CompressStream(opts)
 
+      this.asyncCompress = (opts && typeof(opts.asyncCompress) === 'boolean') ? opts.asyncCompress : false
       Transform.call(this)
 
       // first push the identifier frame
@@ -61,36 +62,66 @@ CompressStream.prototype._uncompressed = function (chunk) {
 /**
  * Some compression benchmarks : 
  * 
- *   i) Chunking in transform with snappy.compressSync (the new implementation)
- *   ii) Chunking from outside with compressStream.write (using original snappy.compress)
+ *   i) Sync compress via snappy.compressSync ({asyncCompress:false}) default
+ *   ii) Async compress via snappy.compress ({asyncCompress:true})
  *   iii) No chunking (Original)
  * 
- *  | Size              | in transform | compressStream.write | orginal (no chunking) |
- *  |-------------------|--------------|----------------------|-----------------------|
- *  | 10kb   (1 chunk)  | 0.0229 ms    | 0.0385 ms            | 0.0388 ms             |
- *  | 100kb  (2 chunks) | 0.0562 ms    | 0.1051 ms            | 0.0844 ms             |
- *  | 1000kb (16 chunks)| 0.382  ms    | 0.7971 ms            | 0.1998 ms             |
+ *  | Size               | sync compress | async compress | original (no chunking) |
+ *  |--------------------|---------------|----------------|------------------------|
+ *  | 10kb   (1 chunk)   | 0.0229 ms     | 0.0385 ms      | 0.0388 ms              |
+ *  | 100kb  (2 chunks)  | 0.0562 ms     | 0.1051 ms      | 0.0844 ms              |
+ *  | 1000kb (16 chunks) | 0.382  ms     | 0.7971 ms      | 0.1998 ms              |
  * 
  */
 
+
 CompressStream.prototype._transform = function(chunk, enc, callback) {
-  new Promise(() => {
+  self = this;
+
+  function asyncCompressNext(startFrom) {
+    const endAt = startFrom + Math.min(chunk.length - startFrom, UNCOMPRESSED_CHUNK_SIZE);
+    const bytesChunk = chunk.slice(startFrom, endAt);
+    snappy.compress(bytesChunk, function(err, compressed) {
+      if (err) {
+        callback(err)
+      } else {
+
+        if (compressed.length < bytesChunk.length)
+          self._compressed(bytesChunk, compressed)
+        else
+          self._uncompressed(bytesChunk)
+
+        if (endAt < chunk.length) {
+          asyncCompressNext(endAt)
+        } else {
+          callback()
+        }
+      }
+    })
+  }
+
+  function syncCompress() {
     try {
       for (let startFrom = 0; startFrom < chunk.length; startFrom += UNCOMPRESSED_CHUNK_SIZE) {
         const endAt = startFrom + Math.min(chunk.length - startFrom, UNCOMPRESSED_CHUNK_SIZE);
         const bytesChunk = chunk.slice(startFrom, endAt);
         const compressed = snappy.compressSync(bytesChunk)
+
         if (compressed.length < bytesChunk.length)
-          this._compressed(bytesChunk, compressed)
+          self._compressed(bytesChunk, compressed)
         else
-          this._uncompressed(bytesChunk)
-
+          self._uncompressed(bytesChunk)
       }
       callback();
     } catch (err) {
       return callback(err);
     }
-  }).catch(e => console.log(e))
+  }
+  if (this.asyncCompress) {
+    asyncCompressNext(0)
+  } else {
+    syncCompress();
+  }
 }
 
 module.exports = CompressStream

test/compress-test.js

@@ -1,79 +1,66 @@
-var spawn = require('child_process').spawn
-  , createCompressStream = require('../').createCompressStream
-  , test = require('tap').test
-  , largerInput = require('fs').readFileSync(__filename)
-  , largerInputString = largerInput.toString()
+const spawn = require('child_process').spawn,
+  createCompressStream = require('../').createCompressStream,
+  test = require('tap').test,
+  largerInput = require('fs').readFileSync(__filename)
 
 const UNCOMPRESSED_CHUNK_SIZE = 65536
 let superLargeInput = largerInput;
 for (let i = largerInput.length; i <= UNCOMPRESSED_CHUNK_SIZE; i += largerInput.length) {
   superLargeInput = Buffer.concat([superLargeInput, largerInput]);
 }
-const superLargeInputString = superLargeInput.toString();
 
-test('compress small string', function (t) {
-  var child = spawn('python', [ '-m', 'snappy', '-d' ])
-    , compressStream = createCompressStream()
-    , data = ''
-
-  child.stdout.on('data', function (chunk) {
-    data = data + chunk.toString()
-  })
-
-  child.stdout.on('end', function () {
-    t.equal(data, 'beep boop')
-    t.end()
-  })
-
-  child.stderr.pipe(process.stderr)
-
-  compressStream.pipe(child.stdin)
-
-  compressStream.write('beep boop')
-  compressStream.end()
-})
-
-test('compress large string', function (t) {
-  var child = spawn('python', [ '-m', 'snappy', '-d' ])
-    , compressStream = createCompressStream()
-    , data = ''
-
-  child.stdout.on('data', function (chunk) {
-    data = data + chunk.toString()
-  })
-
-  child.stdout.on('end', function () {
-    t.equal(data, largerInputString)
-    t.end()
+[{
+  testName: "small",
+  testString: "beep boop",
+  asyncCompress: true
+}, {
+  testName: "small",
+  testString: "beep boop",
+  asyncCompress: false
+}, {
+  testName: "large",
+  testString: largerInput,
+  asyncCompress: true
+}, {
+  testName: "large",
+  testString: largerInput,
+  asyncCompress: false
+}, {
+  testName: "super large",
+  testString: superLargeInput,
+  asyncCompress: true
+}, {
+  testName: "super large",
+  testString: superLargeInput,
+  asyncCompress: false
+}].forEach(({
+  testName,
+  testString,
+  asyncCompress
+}) => {
+
+  test(`compress ${testName} input - asyncCompress=${asyncCompress}`, function(t) {
+    const child = spawn('python', ['-m', 'snappy', '-d']),
+      compressStream = createCompressStream({
+        asyncCompress
+      })
+    let data = ''
+
+    child.stdout.on('data', function(chunk) {
+      data = data + chunk.toString()
+    })
+
+    child.stdout.on('end', function() {
+      t.equal(data, testString.toString())
+      t.end()
+    })
+
+    child.stderr.pipe(process.stderr)
+
+    compressStream.pipe(child.stdin)
+
+    compressStream.write(testString)
+    compressStream.end()
   })
 
-  child.stderr.pipe(process.stderr)
-
-  compressStream.pipe(child.stdin)
-
-  compressStream.write(largerInputString)
-  compressStream.end()
-})
-
-
-test('compress very very large string', function (t) {
-  var child = spawn('python', [ '-m', 'snappy', '-d' ])
-    , compressStream = createCompressStream()
-    , data = ''
-
-  child.stdout.on('data', function (chunk) {
-    data = data + chunk.toString()
-  })
-
-  child.stdout.on('end', function () {
-    t.equal(data, superLargeInputString)
-    t.end()
-  })
-
-  child.stderr.pipe(process.stderr)
-
-  compressStream.pipe(child.stdin)
-
-  compressStream.write(superLargeInputString)
-  compressStream.end()
-})
+})