const { Readable: t, isEnded: s, getStreamError: e } = require("streamx"),
i = require("fast-fifo"),
r = 8
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 }))
}))
})
}
}