stream: fix fromAsyncGen

Fixes: #40497

Author: ronag

lib/internal/streams/duplexify.js

@@ -210,11 +210,13 @@ function fromAsyncGen(fn) {
   const value = fn(async function*() {
     while (true) {
       const { chunk, done, cb } = await promise;
+      promise = null;
+      resolve = null;
       process.nextTick(cb);
       if (done) return;
       if (signal.aborted) throw new AbortError();
-      yield chunk;
       ({ promise, resolve } = createDeferredPromise());
+      yield chunk;
     }
   }(), { signal });
 

test/parallel/test-stream-duplex-from.js

@@ -2,7 +2,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { Duplex, Readable, Writable } = require('stream');
+const { Duplex, Readable, Writable, pipeline } = require('stream');
 
 {
   const d = Duplex.from({
@@ -118,3 +118,34 @@ const { Duplex, Readable, Writable } = require('stream');
     assert.strictEqual(d.readable, false);
   }));
 }
+
+{
+  // https://github.com/nodejs/node/issues/40497
+  pipeline(
+    ['abc\ndef\nghi'],
+    Duplex.from(async function * (source) {
+      let rest = ''
+      for await (const chunk of source) {
+        console.error(0)
+        const lines = (rest + chunk.toString()).split('\n')
+        rest = lines.pop()
+        for (const line of lines) {
+          console.error(1, line)
+          yield line
+          console.error(2)
+        }
+        console.error(3)
+      }
+      console.error(4)
+      yield rest
+    }),
+    async function * (source) {
+      let ret = ''
+      for await (const x of source) {
+        ret += x
+      }
+      assert.strictEqual(ret, 'abcdefghi')
+    },
+    common.mustCall(() => {}),
+  )
+}