stream: readable read one buffer at a time

Instead of wasting cycles concatenating buffers, just return each
one by one.

Author: ronag

lib/internal/streams/readable.js

@@ -631,6 +631,13 @@ function howMuchToRead(n, state) {
   if ((state[kState] & kObjectMode) !== 0)
     return 1;
   if (NumberIsNaN(n)) {
+    // Fast path for buffers.
+    if ((state[kState] & kDecoder) === 0) {
+      return state.length
+        ? state.buffer[state.bufferIndex].length
+        : state.length;
+    }
+
     // Only flow one buffer at a time.
     if ((state[kState] & kFlowing) !== 0 && state.length)
       return state.buffer[state.bufferIndex].length;

test/parallel/test-stream-compose.js

@@ -490,7 +490,7 @@ const assert = require('assert');
 
     newStream.end();
 
-    assert.deepStrictEqual(await newStream.toArray(), [Buffer.from('Steve RogersOn your left')]);
+    assert.deepStrictEqual(await newStream.toArray(), [Buffer.from('Steve Rogers'), Buffer.from('On your left')]);
   })().then(common.mustCall());
 }
 

test/parallel/test-stream-push-strings.js

@@ -59,7 +59,7 @@ ms.on('readable', function() {
     results.push(String(chunk));
 });
 
-const expect = [ 'first chunksecond to last chunk', 'last chunk' ];
+const expect = [ 'first chunk', 'second to last chunk', 'last chunk' ];
 process.on('exit', function() {
   assert.strictEqual(ms._chunks, -1);
   assert.deepStrictEqual(results, expect);

test/parallel/test-stream-readable-emittedReadable.js

@@ -10,7 +10,7 @@ const readable = new Readable({
 // Initialized to false.
 assert.strictEqual(readable._readableState.emittedReadable, false);
 
-const expected = [Buffer.from('foobar'), Buffer.from('quo'), null];
+const expected = [Buffer.from('foo'), Buffer.from('bar'), Buffer.from('quo'), null];
 readable.on('readable', common.mustCall(() => {
   // emittedReadable should be true when the readable event is emitted
   assert.strictEqual(readable._readableState.emittedReadable, true);

test/parallel/test-stream-readable-infinite-read.js

@@ -26,8 +26,8 @@ readable.on('readable', common.mustCall(function() {
   // TODO(mcollina): there is something odd in the highWaterMark logic
   // investigate.
   if (i === 1) {
-    assert.strictEqual(data.length, 8192 * 2);
+    assert.strictEqual(data.length, 8192);
   } else {
-    assert.strictEqual(data.length, 8192 * 3);
+    assert.strictEqual(data.length, 8192);
   }
 }, 11));

test/parallel/test-stream-readable-readable-one.js

@@ -0,0 +1,20 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+
+const { Readable } = require('stream');
+
+// Read one buffer at a time and don't waste cycles allocating
+// and copying into a new larger buffer.
+{
+  const r = new Readable({
+    read() {}
+  });
+  const buffers = [Buffer.allocUnsafe(5), Buffer.allocUnsafe(10)];
+  for (const buf of buffers) {
+    r.push(buf);
+  }
+  for (const buf of buffers) {
+    assert.strictEqual(r.read(), buf);
+  }
+}

test/parallel/test-stream-typedarray.js

@@ -83,9 +83,19 @@ const views = common.getArrayBufferViews(buffer);
   readable.push(views[2]);
   readable.unshift(views[0]);
 
-  const buf = readable.read();
+  let buf;
+
+  buf = readable.read();
+  assert(buf instanceof Buffer);
+  assert.deepStrictEqual([...buf], [...views[0]]);
+
+  buf = readable.read();
+  assert(buf instanceof Buffer);
+  assert.deepStrictEqual([...buf], [...views[1]]);
+
+  buf = readable.read();
   assert(buf instanceof Buffer);
-  assert.deepStrictEqual([...buf], [...views[0], ...views[1], ...views[2]]);
+  assert.deepStrictEqual([...buf], [...views[2]]);
 }
 
 {

test/parallel/test-stream-uint8array.js

@@ -80,9 +80,15 @@ const GHI = new Uint8Array([0x47, 0x48, 0x49]);
   readable.push(DEF);
   readable.unshift(ABC);
 
-  const buf = readable.read();
+  let buf;
+
+  buf = readable.read();
+  assert(buf instanceof Buffer);
+  assert.deepStrictEqual([...buf], [...ABC]);
+
+  buf = readable.read();
   assert(buf instanceof Buffer);
-  assert.deepStrictEqual([...buf], [...ABC, ...DEF]);
+  assert.deepStrictEqual([...buf], [...DEF]);
 }
 
 {

test/parallel/test-stream2-transform.js

@@ -282,7 +282,9 @@ const { PassThrough, Transform } = require('stream');
       pt.write(Buffer.from('ef'), common.mustCall(function() {
         pt.end();
       }));
-      assert.strictEqual(pt.read().toString(), 'abcdef');
+      assert.strictEqual(pt.read().toString(), 'abc');
+      assert.strictEqual(pt.read().toString(), 'd');
+      assert.strictEqual(pt.read().toString(), 'ef');
       assert.strictEqual(pt.read(), null);
     });
   });