fixup! stream: readable read one buffer at a time

Author: ronag

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

@@ -32,12 +32,11 @@ const asyncReadable = new Readable({
 });
 
 asyncReadable.on('readable', common.mustCall(() => {
-  if (asyncReadable.read() !== null) {
-    // After each read(), the buffer is empty.
-    // If the stream doesn't end now,
-    // then we need to notify the reader on future changes.
-    assert.strictEqual(asyncReadable._readableState.needReadable, true);
+  while (asyncReadable.read() !== null) {
   }
+  // If the stream doesn't end now,
+  // then we need to notify the reader on future changes.
+  assert.strictEqual(asyncReadable._readableState.needReadable || asyncReadable._readableState.ended, true);
 }, 2));
 
 process.nextTick(common.mustCall(() => {

test/parallel/test-webstreams-pipeline.js

@@ -126,7 +126,7 @@ const http = require('http');
   });
 
   pipeline(r, ws, common.mustSucceed(() => {
-    assert.deepStrictEqual(values, ['helloworld']);
+    assert.deepStrictEqual(values, ['hello', 'world']);
   }));
 
   r.push('hello');
@@ -181,7 +181,7 @@ const http = require('http');
   });
 
   pipeline(rs, t, ws, common.mustSucceed(() => {
-    assert.deepStrictEqual(values, ['HELLOWORLD']);
+    assert.deepStrictEqual(values, ['HELLO', 'WORLD']);
   }));
 
   c.enqueue('hello');

test/parallel/test-zlib-flush-write-sync-interleaved.js

@@ -19,9 +19,11 @@ for (const chunk of ['abc', 'def', 'ghi']) {
   compress.write(chunk, common.mustCall(() => events.push({ written: chunk })));
   compress.flush(Z_PARTIAL_FLUSH, common.mustCall(() => {
     events.push('flushed');
-    const chunk = compress.read();
-    if (chunk !== null)
+    while (true) {
+      const chunk = compress.read();
+      if (chunk === null) break;
       compressedChunks.push(chunk);
+    }
   }));
 }
 
@@ -36,7 +38,13 @@ function writeToDecompress() {
   const chunk = compressedChunks.shift();
   if (chunk === undefined) return decompress.end();
   decompress.write(chunk, common.mustCall(() => {
-    events.push({ read: decompress.read() });
+    let read = '';
+    while (true) {
+      const x = decompress.read();
+      if (x == null) break;
+      read += x;
+    }
+    if (read) events.push({ read });
     writeToDecompress();
   }));
 }