stream: prevent pipeline hang with generator functions

debadree25 · MoLow · commit d3449ca01001 · 2023-07-07T00:06:09.000+03:00
Fixes: #47708 PR-URL: #47712 Reviewed-By: Robert Nagy <ronagy@icloud.com> Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com> Reviewed-By: Feng Yu <F3n67u@outlook.com>
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -138,10 +138,9 @@ async function pumpToNode(iterable, writable, finish, { end }) {
 
 if (end) {
 writable.end();
+ await wait();
 }
 
- await wait();
-
 finish();
 } catch (err) {
 finish(error !== err ? aggregateTwoErrors(error, err) : err);
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
@@ -1616,3 +1616,21 @@ const tsp = require('timers/promises');
 dup.push(null);
 dup.read();
 }
+
+{
+ let res = '';
+ const writable = new Writable({
+ write(chunk, enc, cb) {
+ res += chunk;
+ cb();
+ }
+ });
+ pipelinep(async function*() {
+ yield 'hello';
+ await Promise.resolve();
+ yield 'world';
+ }, writable, { end: false }).then(common.mustCall(() => {
+ assert.strictEqual(res, 'helloworld');
+ assert.strictEqual(writable.closed, false);
+ }));
+}

-Original file line number
+Diff line change
 if(end){
 writable.end();
 +awaitwait();
+}
 -awaitwait();
+-
 finish();
 }catch(err){
 finish(error!==err ? aggregateTwoErrors(error,err) : err);
-Original file line number
+Diff line change
 dup.push(null);
 dup.read();
+}
++
 +{
 +letres='';
 +constwritable=newWritable({
 +write(chunk,enc,cb){
 +res+=chunk;
 +cb();
 +}
 +});
 +pipelinep(asyncfunction*(){
 +yield'hello';
 +awaitPromise.resolve();
 +yield'world';
 +},writable,{end: false}).then(common.mustCall(()=>{
 +assert.strictEqual(res,'helloworld');
 +assert.strictEqual(writable.closed,false);
 +}));
 +}