stream: preserve half-open duplexes in async iteration

Signed-off-by: Efe Karasakal <hi@efe.dev>
PR-URL: #64275
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Jake Yuesong Li <jake.yuesong@gmail.com>

Author: efekrskl

lib/internal/streams/readable.js

@@ -1401,9 +1401,16 @@ async function* createAsyncIterator(stream, options) {
 error=aggregateTwoErrors(error,err);
 throwerror;
 }finally{
+constpreserveHalfOpenDuplex=
+error===null&&
+stream.allowHalfOpen===true&&
+stream.writable===true&&
+stream.writableEnded!==true;
+
 if(
 (error||options?.destroyOnReturn!==false)&&
-(error===undefined||stream._readableState.autoDestroy)
+(error===undefined||stream._readableState.autoDestroy)&&
+!preserveHalfOpenDuplex
 ){
 destroyImpl.destroyer(stream,null);
 }else{

test/parallel/test-net-allow-half-open-async-iter.js

@@ -0,0 +1,80 @@
+'use strict';
+
+constcommon=require('../common');
+constassert=require('assert');
+constnet=require('net');
+
+(asyncfunction(){
+letresolveServerSocket;
+constserverSocketPromise=newPromise((resolve)=>{
+resolveServerSocket=resolve;
+});
+
+constserver=net.createServer({
+allowHalfOpen: true,
+},common.mustCall((socket)=>{
+resolveServerSocket(socket);
+}));
+
+server.on('error',common.mustNotCall());
+server.on('close',common.mustCall());
+
+awaitnewPromise((resolve)=>{
+server.listen(0,common.localhostIPv4,resolve);
+});
+
+constclientSocket=awaitnewPromise((resolve)=>{
+constsocket=net.createConnection({
+allowHalfOpen: true,
+port: server.address().port,
+host: server.address().address,
+},common.mustCall(()=>{
+resolve(socket);
+}));
+socket.on('error',common.mustNotCall());
+});
+
+constserverSocket=awaitserverSocketPromise;
+serverSocket.on('error',common.mustNotCall());
+
+awaitnewPromise((resolve,reject)=>{
+clientSocket.write('data written to client socket',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+awaitnewPromise((resolve)=>{
+clientSocket.end(resolve);
+});
+
+letserverRead='';
+forawait(constchunkofserverSocket){
+serverRead+=chunk;
+}
+
+assert.strictEqual(serverRead,'data written to client socket');
+assert.strictEqual(serverSocket.destroyed,false);
+
+awaitnewPromise((resolve,reject)=>{
+serverSocket.write('data written to server socket',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+awaitnewPromise((resolve)=>{
+serverSocket.end(resolve);
+});
+
+letclientRead='';
+forawait(constchunkofclientSocket){
+clientRead+=chunk;
+}
+
+assert.strictEqual(clientRead,'data written to server socket');
+
+awaitnewPromise((resolve)=>{
+server.close(resolve);
+});
+})().then(common.mustCall());

test/parallel/test-stream-readable-async-iter-half-open-duplex.js

@@ -0,0 +1,41 @@
+'use strict';
+
+constcommon=require('../common');
+constassert=require('assert');
+const{
+ Duplex,
+}=require('stream');
+
+{
+letwritten='';
+
+constduplex=newDuplex({
+allowHalfOpen: true,
+read(){
+this.push('hello');
+this.push(null);
+},
+write(chunk,encoding,callback){
+written+=chunk;
+callback();
+},
+});
+
+duplex.on('error',common.mustNotCall());
+duplex.on('close',common.mustCall());
+
+(async()=>{
+letread='';
+forawait(constchunkofduplex){
+read+=chunk;
+}
+
+assert.strictEqual(read,'hello');
+assert.strictEqual(duplex.destroyed,false);
+
+duplex.write('world',common.mustSucceed(()=>{
+assert.strictEqual(written,'world');
+duplex.end();
+}));
+})().then(common.mustCall());
+}