stream: use .chunk when calling adapters's writev

Fix: #42157

PR-URL: #42161
Fixes: #42157
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>

Author: meixg

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);
         },

test/parallel/test-whatwg-webstreams-adapters-to-streamduplex.js

@@ -0,0 +1,166 @@
+// Flags: --no-warnings --expose-internals
+'use strict';
+
+const common = require('../common');
+
+const assert = require('assert');
+
+const {
+  TransformStream,
+} = require('stream/web');
+
+const {
+  newStreamDuplexFromReadableWritablePair,
+} = require('internal/webstreams/adapters');
+
+const {
+  finished,
+  pipeline,
+  Readable,
+  Writable,
+} = require('stream');
+
+const {
+  kState,
+} = require('internal/webstreams/util');
+
+{
+  const transform = new TransformStream();
+  const duplex = newStreamDuplexFromReadableWritablePair(transform);
+
+  assert(transform.readable.locked);
+  assert(transform.writable.locked);
+
+  duplex.destroy();
+
+  duplex.on('close', common.mustCall(() => {
+    assert.strictEqual(transform.readable[kState].state, 'closed');
+    assert.strictEqual(transform.writable[kState].state, 'errored');
+  }));
+}
+
+{
+  const error = new Error('boom');
+  const transform = new TransformStream();
+  const duplex = newStreamDuplexFromReadableWritablePair(transform);
+
+  assert(transform.readable.locked);
+  assert(transform.writable.locked);
+
+  duplex.destroy(error);
+  duplex.on('error', common.mustCall((reason) => {
+    assert.strictEqual(reason, error);
+  }));
+
+  duplex.on('close', common.mustCall(() => {
+    assert.strictEqual(transform.readable[kState].state, 'closed');
+    assert.strictEqual(transform.writable[kState].state, 'errored');
+    assert.strictEqual(transform.writable[kState].storedError, error);
+  }));
+}
+
+{
+  const transform = new TransformStream();
+  const duplex = new newStreamDuplexFromReadableWritablePair(transform);
+
+  duplex.end();
+  duplex.resume();
+
+  duplex.on('close', common.mustCall(() => {
+    assert.strictEqual(transform.readable[kState].state, 'closed');
+    assert.strictEqual(transform.writable[kState].state, 'closed');
+  }));
+}
+
+{
+  const ec = new TextEncoder();
+  const dc = new TextDecoder();
+  const transform = new TransformStream({
+    transform(chunk, controller) {
+      const text = dc.decode(chunk);
+      controller.enqueue(ec.encode(text.toUpperCase()));
+    }
+  });
+  const duplex = new newStreamDuplexFromReadableWritablePair(transform, {
+    encoding: 'utf8',
+  });
+
+  duplex.end('hello');
+  duplex.on('data', common.mustCall((chunk) => {
+    assert.strictEqual(chunk, 'HELLO');
+  }));
+  duplex.on('end', common.mustCall());
+
+  duplex.on('close', common.mustCall(() => {
+    assert.strictEqual(transform.readable[kState].state, 'closed');
+    assert.strictEqual(transform.writable[kState].state, 'closed');
+  }));
+}
+
+{
+  const ec = new TextEncoder();
+  const dc = new TextDecoder();
+  const transform = new TransformStream({
+    transform: common.mustCall((chunk, controller) => {
+      const text = dc.decode(chunk);
+      controller.enqueue(ec.encode(text.toUpperCase()));
+    })
+  });
+  const duplex = new newStreamDuplexFromReadableWritablePair(transform, {
+    encoding: 'utf8',
+  });
+
+  finished(duplex, common.mustCall());
+
+  duplex.end('hello');
+  duplex.resume();
+}
+
+{
+  const ec = new TextEncoder();
+  const dc = new TextDecoder();
+  const transform = new TransformStream({
+    transform: common.mustCall((chunk, controller) => {
+      const text = dc.decode(chunk);
+      controller.enqueue(ec.encode(text.toUpperCase()));
+    })
+  });
+  const duplex = new newStreamDuplexFromReadableWritablePair(transform, {
+    encoding: 'utf8',
+  });
+
+  const readable = new Readable({
+    read() {
+      readable.push(Buffer.from('hello'));
+      readable.push(null);
+    }
+  });
+
+  const writable = new Writable({
+    write: common.mustCall((chunk, encoding, callback) => {
+      assert.strictEqual(dc.decode(chunk), 'HELLO');
+      assert.strictEqual(encoding, 'buffer');
+      callback();
+    })
+  });
+
+  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();
+}