stream: when value is already in buffer don't emit the next one

fix 46765
nodejs · Feb 24, 2023 · 426eb7c · 426eb7c
1 parent 03f18ab
commit 426eb7c
Show file tree

Hide file tree

Showing 2 changed files with 71 additions and 7 deletions.
diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js
@@ -366,7 +366,7 @@ Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) {
 // in the queue, and wait our turn.  Otherwise, call _write
 // If we return false, then we need a drain event, so set that flag.
 function writeOrBuffer(stream, state, chunk, encoding, callback) {
-  const len = state.objectMode ? 1 : chunk.length;
+  let len = state.objectMode ? 1 : chunk.length;
 
   state.length += len;
 
@@ -385,12 +385,19 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
       state.allNoop = false;
     }
   } else {
-    state.writelen = len;
-    state.writecb = callback;
-    state.writing = true;
-    state.sync = true;
-    stream._write(chunk, encoding, state.onwrite);
-    state.sync = false;
+    if(state.buffered.length) {
+      state.buffered.push({ chunk, encoding, callback });
+      const firstBuffered = state.buffered.shift();
+      len = state.objectMode ? 1 : firstBuffered.chunk.length;
+      chunk = firstBuffered.chunk;
+      callback = firstBuffered.callback;
+    }
+      state.writelen = len;
+      state.writecb = callback;
+      state.writing = true;
+      state.sync = true;
+      stream._write(chunk, encoding, state.onwrite);
+      state.sync = false;
   }
 
   // Return false if errored or destroyed in order to break

diff --git a/test/parallel/test-stream2-transform.js b/test/parallel/test-stream2-transform.js
@@ -23,6 +23,7 @@
 const common = require('../common');
 const assert = require('assert');
 const { PassThrough, Transform } = require('stream');
+const { Readable } = require('node:stream');
 
 {
   // Verify writable side consumption
@@ -468,3 +469,59 @@ const { PassThrough, Transform } = require('stream');
     assert.strictEqual(ended, true);
   }));
 }
+
+{
+  const createInnerTransform = () => new Transform({
+    objectMode: true,
+
+    construct(callback) {
+      this.push('header from constructor');
+      callback();
+    },
+
+    transform: (row, encoding, callback) => {
+      callback(null, 'transform | ' + row);
+    },
+  });
+
+  const createOuterTransform = () => {
+    let innerTransform;
+
+    return new Transform({
+      objectMode: true,
+
+      transform(row, encoding, callback) {
+        if (!innerTransform) {
+          innerTransform = createInnerTransform();
+          innerTransform.on('data', (data) => {
+            this.push(data);
+          });
+
+          callback();
+        } else if (innerTransform.write('outer | ' + row)) {
+          process.nextTick(callback);
+        } else {
+          innerTransform.once('drain', callback);
+        }
+      },
+    });
+  };
+
+  Readable.from([
+      'create InnerTransform',
+      'firstLine',
+      'secondLine',
+    ])
+      .compose(createOuterTransform())
+    .toArray().then(
+    common.mustCall((value) => {
+      assert.deepStrictEqual(value, [
+        'header from constructor',
+        'transform | outer | firstLine',
+        'transform | outer | secondLine',
+      ]);
+    }),
+  );
+
+
+}