stream: use .chunk when calling adapters's writev

Fix: #42157 PR-URL: #42161 Fixes: #42157 Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
nodejs · Mar 3, 2022 · 54bc3c9 · 54bc3c9
1 parent 010cb71
commit 54bc3c9
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 5 deletions.
diff --git a/lib/internal/webstreams/adapters.js b/lib/internal/webstreams/adapters.js
@@ -228,8 +228,9 @@ function newStreamWritableFromWritableStream(writableStream, options = {}) {
 
     writev(chunks, callback) {
       function done(error) {
+        error = error.filter((e) => e);
         try {
-          callback(error);
+          callback(error.length === 0 ? undefined : error);
         } catch (error) {
           // In a next tick because this is happening within
           // a promise context, and if there are any errors
@@ -247,7 +248,7 @@ function newStreamWritableFromWritableStream(writableStream, options = {}) {
             PromiseAll(
               ArrayPrototypeMap(
                 chunks,
-                (chunk) => writer.write(chunk))),
+                (data) => writer.write(data.chunk))),
             done,
             done);
         },
@@ -633,8 +634,9 @@ function newStreamDuplexFromReadableWritablePair(pair = {}, options = {}) {
 
     writev(chunks, callback) {
       function done(error) {
+        error = error.filter((e) => e);
         try {
-          callback(error);
+          callback(error.length === 0 ? undefined : error);
         } catch (error) {
           // In a next tick because this is happening within
           // a promise context, and if there are any errors
@@ -652,7 +654,7 @@ function newStreamDuplexFromReadableWritablePair(pair = {}, options = {}) {
             PromiseAll(
               ArrayPrototypeMap(
                 chunks,
-                (chunk) => writer.write(chunk))),
+                (data) => writer.write(data.chunk))),
             done,
             done);
         },

diff --git a/test/parallel/test-whatwg-webstreams-adapters-to-streamduplex.js b/test/parallel/test-whatwg-webstreams-adapters-to-streamduplex.js
@@ -147,3 +147,20 @@ const {
   finished(duplex, common.mustCall());
   pipeline(readable, duplex, writable, common.mustCall());
 }
+
+{
+  const transform = new TransformStream();
+  const duplex = newStreamDuplexFromReadableWritablePair(transform);
+  duplex.setEncoding('utf-8');
+  duplex.on('data', common.mustCall((data) => {
+    assert.strictEqual(data, 'hello');
+  }, 5));
+
+  duplex.write(Buffer.from('hello'));
+  duplex.write(Buffer.from('hello'));
+  duplex.write(Buffer.from('hello'));
+  duplex.write(Buffer.from('hello'));
+  duplex.write(Buffer.from('hello'));
+
+  duplex.end();
+}
diff --git a/test/parallel/test-whatwg-webstreams-adapters-to-streamwritable.js b/test/parallel/test-whatwg-webstreams-adapters-to-streamwritable.js
@@ -200,14 +200,17 @@ class TestSource {
 
 {
   const writableStream = new WritableStream({
-    write: common.mustCall(2),
+    write: common.mustCall(5),
     close: common.mustCall(),
   });
   const writable = newStreamWritableFromWritableStream(writableStream);
 
   finished(writable, common.mustCall());
 
   writable.write('hello');
+  writable.write('hello');
+  writable.write('hello');
+  writable.write('world');
   writable.write('world');
   writable.end();
 }