{
  "version": 3,
  "sources": ["../src/queueless-pushable.ts"],
  "sourcesContent": ["/**\n * @packageDocumentation\n *\n * A pushable async generator that waits until the current value is consumed\n * before allowing a new value to be pushed.\n *\n * Useful for when you don't want to keep memory usage under control and/or\n * allow a downstream consumer to dictate how fast data flows through a pipe,\n * but you want to be able to apply a transform to that data.\n *\n * @example\n *\n * ```typescript\n * import { queuelessPushable } from 'it-queueless-pushable'\n *\n * const pushable = queuelessPushable<string>()\n *\n * // run asynchronously\n * Promise.resolve().then(async () => {\n *   // push a value - the returned promise will not resolve until the value is\n *   // read from the pushable\n *   await pushable.push('hello')\n * })\n *\n * // read a value\n * const result = await pushable.next()\n * console.info(result) // { done: false, value: 'hello' }\n * ```\n */\n\nimport deferred from 'p-defer'\nimport { raceSignal } from 'race-signal'\nimport type { AbortOptions } from 'abort-error'\nimport type { RaceSignalOptions } from 'race-signal'\n\nexport interface Pushable<T> extends AsyncGenerator<T, void, unknown> {\n  /**\n   * End the iterable after all values in the buffer (if any) have been yielded. If an\n   * error is passed the buffer is cleared immediately and the next iteration will\n   * throw the passed error\n   */\n  end(err?:Error, options?:AbortOptions & RaceSignalOptions):Promise<void>\n\n  /**\n   * Push a value into the iterable. Values are yielded from the iterable in the order\n   * they are pushed. Values not yet consumed from the iterable are buffered.\n   */\n  push(value:T, options?:AbortOptions & RaceSignalOptions):Promise<void>\n}\n\nclass QueuelessPushable <T> implements Pushable<T> {\n    private readNext:PromiseWithResolvers<void>\n    private haveNext:PromiseWithResolvers<void>\n    private ended:boolean\n    private nextResult:IteratorResult<T> | undefined\n    private error?:Error\n\n    constructor () {\n        this.ended = false\n\n        this.readNext = deferred()\n        this.haveNext = deferred()\n    }\n\n    async [Symbol.asyncDispose] ():Promise<void> {\n        await this.end()\n    }\n\n    [Symbol.asyncIterator] ():AsyncGenerator<T, void, unknown> {\n        return this\n    }\n\n    async next ():Promise<IteratorResult<T, void>> {\n        if (this.nextResult == null) {\n            // wait for the supplier to push a value\n            await this.haveNext.promise\n        }\n\n        if (this.nextResult == null) {\n            throw new Error('HaveNext promise resolved but nextResult was undefined')\n        }\n\n        const nextResult = this.nextResult\n        this.nextResult = undefined\n\n        // signal to the supplier that we read the value\n        this.readNext.resolve()\n        this.readNext = deferred()\n\n        return nextResult\n    }\n\n    async throw (err?:Error):Promise<IteratorReturnResult<undefined>> {\n        this.ended = true\n        this.error = err\n\n        if (err != null) {\n            // this can cause unhandled promise rejections if nothing is awaiting the\n            // next value so attach a dummy catch listener to the promise\n            this.haveNext.promise.catch(() => {})\n            this.haveNext.reject(err)\n        }\n\n        const result:IteratorReturnResult<undefined> = {\n            done: true,\n            value: undefined\n        }\n\n        return result\n    }\n\n    async return ():Promise<IteratorResult<T>> {\n        const result:IteratorReturnResult<undefined> = {\n            done: true,\n            value: undefined\n        }\n\n        this.ended = true\n        this.nextResult = result\n\n        // let the consumer know we have a new value\n        this.haveNext.resolve()\n\n        return result\n    }\n\n    async push (value:T, options?:AbortOptions & RaceSignalOptions):Promise<void> {\n        await this._push(value, options)\n    }\n\n    async end (err?:Error, options?:AbortOptions & RaceSignalOptions):Promise<void> {\n        if (err != null) {\n            await this.throw(err)\n        } else {\n            // abortable return\n            await this._push(undefined, options)\n        }\n    }\n\n    private async _push (value?:T, options?:AbortOptions & RaceSignalOptions):Promise<void> {\n        if (value != null && this.ended) {\n            throw this.error ?? new Error('Cannot push value onto an ended pushable')\n        }\n\n        // wait for all values to be read\n        while (this.nextResult != null) {\n            await this.readNext.promise\n        }\n\n        if (value != null) {\n            this.nextResult = { done: false, value }\n        } else {\n            this.ended = true\n            this.nextResult = { done: true, value: undefined }\n        }\n\n        // let the consumer know we have a new value\n        this.haveNext.resolve()\n        this.haveNext = deferred()\n\n        // wait for the consumer to have finished processing the value and requested\n        // the next one or for the passed signal to abort the waiting\n        await raceSignal(\n            this.readNext.promise,\n            options?.signal,\n            options\n        )\n    }\n}\n\nexport function queuelessPushable <T> ():Pushable<T> {\n    return new QueuelessPushable<T>()\n}\n"],
  "mappings": ";;;;;;;;;;;;;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AA8BA,qBAAqB;AACrB,yBAA2B;AAmB3B,MAAM,kBAA6C;AAAA,EAlDnD,OAkDmD;AAAA;AAAA;AAAA,EACvC;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EAER,cAAe;AACX,SAAK,QAAQ;AAEb,SAAK,eAAW,eAAAA,SAAS;AACzB,SAAK,eAAW,eAAAA,SAAS;AAAA,EAC7B;AAAA,EAEA,OAAO,OAAO,YAAY,IAAmB;AACzC,UAAM,KAAK,IAAI;AAAA,EACnB;AAAA,EAEA,CAAC,OAAO,aAAa,IAAsC;AACvD,WAAO;AAAA,EACX;AAAA,EAEA,MAAM,OAAyC;AAC3C,QAAI,KAAK,cAAc,MAAM;AAEzB,YAAM,KAAK,SAAS;AAAA,IACxB;AAEA,QAAI,KAAK,cAAc,MAAM;AACzB,YAAM,IAAI,MAAM,wDAAwD;AAAA,IAC5E;AAEA,UAAM,aAAa,KAAK;AACxB,SAAK,aAAa;AAGlB,SAAK,SAAS,QAAQ;AACtB,SAAK,eAAW,eAAAA,SAAS;AAEzB,WAAO;AAAA,EACX;AAAA,EAEA,MAAM,MAAO,KAAqD;AAC9D,SAAK,QAAQ;AACb,SAAK,QAAQ;AAEb,QAAI,OAAO,MAAM;AAGb,WAAK,SAAS,QAAQ,MAAM,MAAM;AAAA,MAAC,CAAC;AACpC,WAAK,SAAS,OAAO,GAAG;AAAA,IAC5B;AAEA,UAAM,SAAyC;AAAA,MAC3C,MAAM;AAAA,MACN,OAAO;AAAA,IACX;AAEA,WAAO;AAAA,EACX;AAAA,EAEA,MAAM,SAAqC;AACvC,UAAM,SAAyC;AAAA,MAC3C,MAAM;AAAA,MACN,OAAO;AAAA,IACX;AAEA,SAAK,QAAQ;AACb,SAAK,aAAa;AAGlB,SAAK,SAAS,QAAQ;AAEtB,WAAO;AAAA,EACX;AAAA,EAEA,MAAM,KAAM,OAAS,SAAyD;AAC1E,UAAM,KAAK,MAAM,OAAO,OAAO;AAAA,EACnC;AAAA,EAEA,MAAM,IAAK,KAAY,SAAyD;AAC5E,QAAI,OAAO,MAAM;AACb,YAAM,KAAK,MAAM,GAAG;AAAA,IACxB,OAAO;AAEH,YAAM,KAAK,MAAM,QAAW,OAAO;AAAA,IACvC;AAAA,EACJ;AAAA,EAEA,MAAc,MAAO,OAAU,SAAyD;AACpF,QAAI,SAAS,QAAQ,KAAK,OAAO;AAC7B,YAAM,KAAK,SAAS,IAAI,MAAM,0CAA0C;AAAA,IAC5E;AAGA,WAAO,KAAK,cAAc,MAAM;AAC5B,YAAM,KAAK,SAAS;AAAA,IACxB;AAEA,QAAI,SAAS,MAAM;AACf,WAAK,aAAa,EAAE,MAAM,OAAO,MAAM;AAAA,IAC3C,OAAO;AACH,WAAK,QAAQ;AACb,WAAK,aAAa,EAAE,MAAM,MAAM,OAAO,OAAU;AAAA,IACrD;AAGA,SAAK,SAAS,QAAQ;AACtB,SAAK,eAAW,eAAAA,SAAS;AAIzB,cAAM;AAAA,MACF,KAAK,SAAS;AAAA,MACd,SAAS;AAAA,MACT;AAAA,IACJ;AAAA,EACJ;AACJ;AAEO,SAAS,oBAAqC;AACjD,SAAO,IAAI,kBAAqB;AACpC;AAFgB;",
  "names": ["deferred"]
}