stream: add tests for Duplex.from()

nodejs · Sep 24, 2024 · 2c9548a · 2c9548a
1 parent 12439c6
commit 2c9548a
Show file tree

Hide file tree

Showing 2 changed files with 57 additions and 3 deletions.
diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
@@ -257,9 +257,6 @@ function fromAsyncGen(fn, destructor) {
   asyncGenerator.throw = async function(err) {
     try {
       return await originalThrow.call(this, err);
-    } catch (error) {
-      err ??= error;
-      throw error;
     } finally {
       if (promise) {
         const _promise = promise;

diff --git a/test/parallel/test-stream-duplex-from.js b/test/parallel/test-stream-duplex-from.js
@@ -507,6 +507,48 @@ function makeATestWritableStream(writeFunc) {
   );
 }
 
+{
+  const r = Readable.from(['foo', 'bar', 'baz']);
+  const d = Duplex.from(async function(asyncGenerator) {
+    while (!(await asyncGenerator.next()).done) await sleep(100);
+  });
+
+  setTimeout(() => d.destroy(), 150);
+
+  pipeline(
+    r,
+    d,
+    common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
+      assert.strictEqual(r.destroyed, true);
+    })
+  );
+}
+
+{
+  const r = Duplex.from(async function* () {
+    for (const value of ['foo', 'bar', 'baz']) {
+      await sleep(50);
+      yield value;
+    }
+  });
+  const d = Duplex.from(async function(asyncGenerator) {
+    while (!(await asyncGenerator.next()).done);
+  });
+
+  setTimeout(() => r.destroy(), 75);
+
+  pipeline(
+    r,
+    d,
+    common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
+      assert.strictEqual(r.destroyed, true);
+      assert.strictEqual(d.destroyed, true);
+    })
+  );
+}
+
 {
   const r = Readable.from(['foo']);
   pipeline(
@@ -535,3 +577,18 @@ function makeATestWritableStream(writeFunc) {
     })
   );
 }
+
+{
+  const r = Readable.from(['foo', 'bar']);
+  pipeline(
+    r,
+    Duplex.from(async function(asyncGenerator) {
+      await asyncGenerator.next();
+      await asyncGenerator.throw();
+    }),
+    common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ABORT_ERR');
+      assert.strictEqual(r.destroyed, true);
+    })
+  );
+}