Streaming support on Node

src/crypto/hash/index.js

@@ -30,8 +30,9 @@ const Buffer = util.getNodeBuffer();
 function node_hash(type) {
   return function (data) {
     const shasum = nodeCrypto.createHash(type);
-    shasum.update(new Buffer(data));
-    return new Uint8Array(shasum.digest());
+    return stream.transform(data, value => {
+      shasum.update(new Buffer(value));
+    }, () => new Uint8Array(shasum.digest()));
   };
 }
 

src/packet/compressed.js

@@ -133,7 +133,7 @@ const nodeZlib = util.getNodeZlib();
 
 function node_zlib(func, options = {}) {
   return function (data) {
-    return func(data, options);
+    return stream.nodeToWeb(stream.webToNode(data).pipe(func(options)));
   };
 }
 
@@ -163,17 +163,17 @@ let decompress_fns;
 if (nodeZlib) { // Use Node native zlib for DEFLATE compression/decompression
   compress_fns = {
     // eslint-disable-next-line no-sync
-    zip: node_zlib(nodeZlib.deflateRawSync, { level: config.deflate_level }),
+    zip: node_zlib(nodeZlib.createDeflateRaw, { level: config.deflate_level }),
     // eslint-disable-next-line no-sync
-    zlib: node_zlib(nodeZlib.deflateSync, { level: config.deflate_level }),
+    zlib: node_zlib(nodeZlib.createDeflate, { level: config.deflate_level }),
     bzip2: bzip2(Bzip2.compressFile)
   };
 
   decompress_fns = {
     // eslint-disable-next-line no-sync
-    zip: node_zlib(nodeZlib.inflateRawSync),
+    zip: node_zlib(nodeZlib.createInflateRaw),
     // eslint-disable-next-line no-sync
-    zlib: node_zlib(nodeZlib.inflateSync),
+    zlib: node_zlib(nodeZlib.createInflate),
     bzip2: bzip2(Bzip2.decompressFile)
   };
 } else { // Use JS fallbacks

src/packet/sym_encrypted_integrity_protected.js

@@ -172,19 +172,16 @@ function aesDecrypt(algo, ct, key) {
   return stream.subarray(pt, crypto.cipher[algo].blockSize + 2); // Remove random prefix
 }
 
-function nodeEncrypt(algo, prefix, pt, key) {
+function nodeEncrypt(algo, pt, key) {
   key = new Buffer(key);
   const iv = new Buffer(new Uint8Array(crypto.cipher[algo].blockSize));
   const cipherObj = new nodeCrypto.createCipheriv('aes-' + algo.substr(3, 3) + '-cfb', key, iv);
-  const ct = cipherObj.update(new Buffer(util.concat([prefix, pt])));
-  return new Uint8Array(ct);
+  return stream.transform(pt, value => new Uint8Array(cipherObj.update(new Buffer(value))));
 }
 
 function nodeDecrypt(algo, ct, key) {
-  ct = new Buffer(ct);
   key = new Buffer(key);
   const iv = new Buffer(new Uint8Array(crypto.cipher[algo].blockSize));
   const decipherObj = new nodeCrypto.createDecipheriv('aes-' + algo.substr(3, 3) + '-cfb', key, iv);
-  const pt = decipherObj.update(ct);
-  return new Uint8Array(pt);
+  return stream.transform(ct, value => new Uint8Array(decipherObj.update(new Buffer(value))));
 }

src/stream.js

@@ -4,6 +4,8 @@ if (typeof ReadableStream === 'undefined') {
   Object.assign(typeof window !== 'undefined' ? window : global, require('web-streams-polyfill'));
 }
 
+const nodeStream = util.getNodeStream();
+
 function concat(arrays) {
   const readers = arrays.map(getReader);
   let current = 0;
@@ -104,7 +106,76 @@ async function readToEnd(input, join) {
 }
 
 
-export default { concat, getReader, transform, clone, subarray, readToEnd };
+/**
+ * Web / node stream conversion functions
+ * From https://github.com/gwicke/node-web-streams
+ */
+
+let nodeToWeb;
+let webToNode;
+
+if (nodeStream) {
+
+  nodeToWeb = function(nodeStream) {
+    return new ReadableStream({
+      start(controller) {
+        nodeStream.pause();
+        nodeStream.on('data', chunk => {
+          controller.enqueue(chunk);
+          nodeStream.pause();
+        });
+        nodeStream.on('end', () => controller.close());
+        nodeStream.on('error', e => controller.error(e));
+      },
+      pull() {
+        nodeStream.resume();
+      },
+      cancel() {
+        nodeStream.pause();
+      }
+    });
+  };
+
+
+  class NodeReadable extends nodeStream.Readable {
+    constructor(webStream, options) {
+      super(options);
+      this._webStream = webStream;
+      this._reader = getReader(webStream);
+      this._reading = false;
+    }
+
+    _read(size) {
+      if (this._reading) {
+        return;
+      }
+      this._reading = true;
+      const doRead = () => {
+        this._reader.read()
+          .then(res => {
+            if (res.done) {
+              this.push(null);
+              return;
+            }
+            if (this.push(res.value)) {
+              return doRead(size);
+            } else {
+              this._reading = false;
+            }
+          });
+      };
+      doRead();
+    }
+  }
+
+  webToNode = function(webStream) {
+    return new NodeReadable(webStream);
+  };
+
+}
+
+
+export default { concat, getReader, transform, clone, subarray, readToEnd, nodeToWeb, webToNode };
 
 
 /*const readerAcquiredMap = new Map();

src/util.js

@@ -630,7 +630,7 @@ export default {
     // This "hack" allows us to access the native node buffer module.
     // otherwise, it gets replaced with the browserified version
     // eslint-disable-next-line no-useless-concat, import/no-dynamic-require
-    return require('buf'+'fer').Buffer;
+    return require('buffer'+'').Buffer;
   },
 
   getNodeZlib: function() {
@@ -641,6 +641,17 @@ export default {
     return require('zlib');
   },
 
+  getNodeStream: function() {
+    if (!util.detectNode()) {
+      return;
+    }
+
+    // This "hack" allows us to access the native node buffer module.
+    // otherwise, it gets replaced with the browserified version
+    // eslint-disable-next-line no-useless-concat, import/no-dynamic-require
+    return require('stream'+'');
+  },
+
   isEmailAddress: function(data) {
     if (!util.isString(data)) {
       return false;