stream: add Readable.readv

A faster alternative to Readable.read for certain use cases.

e.g.

fs.writevSync(fd, readable.readv())

Author: ronag

lib/internal/streams/readable.js

@@ -641,8 +641,16 @@ function howMuchToRead(n, state) {
   return (state[kState] & kEnded) !== 0 ? state.length : 0;
 }
 
+Readable.prototype.readv = function readv () {
+  return _read.call(this, true);
+};
+
+Readable.prototype.read = function read () {
+  return _read.call(this, false);
+}
+
 // You can override either this method, or the async _read(n) below.
-Readable.prototype.read = function(n) {
+function _read (n, returnArr) {
   debug('read', n);
   // Same as parseInt(undefined, 10), however V8 7.3 performance regressed
   // in this scenario, so we are doing it manually.
@@ -748,7 +756,7 @@ Readable.prototype.read = function(n) {
 
   let ret;
   if (n > 0)
-    ret = fromList(n, state);
+    ret = fromList(n, state, returnArr);
   else
     ret = null;
 
@@ -777,7 +785,13 @@ Readable.prototype.read = function(n) {
 
   if (ret !== null && (state[kState] & (kErrorEmitted | kCloseEmitted)) === 0) {
     state[kState] |= kDataEmitted;
-    this.emit('data', ret);
+    if (returnArr) {
+      for (let i = 0; i < ret.length; ++i) {
+        this.emit('data', ret[i]);
+      }
+    } else {
+      this.emit('data', ret);
+    }
   }
 
   return ret;
@@ -1582,7 +1596,7 @@ Readable._fromList = fromList;
 // Length is the combined lengths of all the buffers in the list.
 // This function is designed to be inlinable, so please take care when making
 // changes to the function body.
-function fromList(n, state) {
+function fromList(n, state, returnArr) {
   // nothing buffered.
   if (state.length === 0)
     return null;
@@ -1594,8 +1608,13 @@ function fromList(n, state) {
   const len = buf.length;
 
   if ((state[kState] & kObjectMode) !== 0) {
-    ret = buf[idx];
-    buf[idx++] = null;
+    if (returnArr) {
+      ret = buf.slice(0, idx);
+    } else {
+      ret = buf[idx];
+      buf[idx++] = null;
+      ret = returnArr ? [ret] : ret;
+    }
   } else if (!n || n >= state.length) {
     // Read it all, truncate the list.
     if ((state[kState] & kDecoder) !== 0) {
@@ -1604,11 +1623,16 @@ function fromList(n, state) {
         ret += buf[idx];
         buf[idx++] = null;
       }
+      ret = returnArr ? [ret] : ret;
     } else if (len - idx === 0) {
       ret = Buffer.alloc(0);
+      ret = returnArr ? [ret] : ret;
     } else if (len - idx === 1) {
       ret = buf[idx];
       buf[idx++] = null;
+      ret = returnArr ? [ret] : ret;
+    } else if (returnArr) {
+      ret = buf.slice(0, idx);
     } else {
       ret = Buffer.allocUnsafe(state.length);
 
@@ -1623,10 +1647,12 @@ function fromList(n, state) {
     // `slice` is the same for buffers and strings.
     ret = buf[idx].slice(0, n);
     buf[idx] = buf[idx].slice(n);
+    ret = returnArr ? [ret] : ret;
   } else if (n === buf[idx].length) {
     // First chunk is a perfect match.
     ret = buf[idx];
     buf[idx++] = null;
+    ret = returnArr ? [ret] : ret;
   } else if ((state[kState] & kDecoder) !== 0) {
     ret = '';
     while (idx < len) {
@@ -1646,6 +1672,27 @@ function fromList(n, state) {
         break;
       }
     }
+    ret = returnArr ? [ret] : ret;
+  } else if (returnArr) {
+    ret = [];
+    const retLen = n;
+    while (idx < len) {
+      const data = buf[idx];
+      if (n > data.length) {
+        ret.push(data);
+        n -= data.length;
+        buf[idx++] = null;
+      } else {
+        if (n === data.length) {
+          ret.push(data);
+          buf[idx++] = null;
+        } else {
+          ret.push(new FastBuffer(data.buffer, data.byteOffset, n));
+          buf[idx] = new FastBuffer(data.buffer, data.byteOffset + n, data.length - n);
+        }
+        break;
+      }
+    }
   } else {
     ret = Buffer.allocUnsafe(n);