const{Readable:t,isEnded:s,getStreamError:e}=require("streamx"),i=require("fast-fifo");module.exports=class extends t{constructor(t,s,e){super(e),this.stream=t,this.map=s,this.queue=new i,this.draining=!1,this._continueDestroy=null,this._streamClosed=!1}_open(t){this.stream.on("data",this._ondata.bind(this)),this.stream.on("end",this._drain.bind(this)),this.stream.on("error",this.destroy.bind(this)),this.stream.on("close",this._onclose.bind(this)),t()}_read(t){this._drain(),t()}_destroy(t){this.stream.destroy(),this._continueDestroy=t,this._continueDestroyMaybe()}_continueDestroyMaybe(){if(!this._streamClosed)return;if(this.draining)return;if(!this._continueDestroy)return;const t=this._continueDestroy;this._continueDestroy=null,t()}_onclose(t){this._streamClosed=!0,s(this.stream)||this.destroy(),this._continueDestroyMaybe()}_ondata(t){if(this.destroying)return;const s=this.map(t);this.queue.push(s)>=8&&this.stream.pause(),this._drain()}async _drain(){if(!this.draining){for(this.draining=!0;this.queue.length&&!t.isBackpressured(this);){const t=this.queue.shift();try{this.push(await t)}catch(t){return void this.destroy(t)}this.queue.length<8&&this.stream.resume()}this.draining=!1,s(this.stream)&&this.push(null),this._continueDestroyMaybe()}}async toArray(){const t=this,i=[];return new Promise((r,n)=>{t.on("error",()=>{}),t.on("readable",function(){for(;;){const s=t.read();if(null===s)return;i.push(s)}}),t.on("close",function(){s(t)?r(i):n(e(t,{all:!0}))})})}};