{"version":3,"sources":["../src/sequence.ts"],"sourcesContent":["import { Async } from './async.js';\nimport { Signal } from './signal.js';\nimport { RingBuffer } from './ring-buffer.js';\n\n/**\n * A sequence is a FIFO (First-In-First-Out) queue for async consumption.\n * Designed for single consumer with multiple producers pattern.\n * Values are queued and consumed in order, with backpressure support.\n * Respects an optional AbortSignal: emit() returns false when aborted; waits reject.\n *\n * Key characteristics:\n * - Single consumer - values are consumed once, in order\n * - Multiple producers can push values concurrently\n * - FIFO ordering - first value in is first value out\n * - Backpressure control via reserve() method\n * - Async iteration support for continuous consumption\n *\n * @template T The type of values in the sequence.\n *\n * @example\n * ```typescript\n * // Create a sequence for processing tasks\n * const tasks = new Sequence<string>();\n *\n * // Producer: Add tasks to the queue\n * tasks.emit('task1');\n * tasks.emit('task2');\n * tasks.emit('task3');\n *\n * // Consumer: Process tasks in order\n * const task1 = await tasks.receive(); // 'task1'\n * const task2 = await tasks.receive(); // 'task2'\n * const task3 = await tasks.receive(); // 'task3'\n * ```\n */\nexport class Sequence<T> extends Async<T, boolean> {\n  #queue: RingBuffer<T>;\n  #nextSignal: Signal<void>;\n  #sendSignal: Signal<void>;\n\n  readonly [Symbol.toStringTag] = 'Sequence';\n\n  /**\n   * Merges multiple source sequences into a target sequence.\n   * Values from all sources are forwarded to the target sequence.\n   * Each source is consumed independently and concurrently.\n   *\n   * @param target The sequence that will receive values from all sources\n   * @param sequences The source sequences to merge from\n   *\n   * @example\n   * ```typescript\n   * // Create target and source sequences\n   * const target = new Sequence<number>();\n   * const source1 = new Sequence<number>();\n   * const source2 = new Sequence<number>();\n   *\n   * // Merge sources into target\n   * Sequence.merge(target, source1, source2);\n   *\n   * // Values from both sources appear in target\n   * source1.emit(1);\n   * source2.emit(2);\n   * source1.emit(3);\n   *\n   * // Consumer gets values as they arrive\n   * await target.receive(); // Could be 1, 2, or 3 depending on timing\n   * ```\n   */\n  static merge<T>(target: Sequence<T>, ...sequences: Sequence<T>[]): void {\n    for (const source of sequences) {\n      queueMicrotask(async () => {\n        if (!target.disposed)\n          try {\n            const sink = target.sink;\n            for await (const value of source) {\n              if (!sink(value)) {\n                return;\n              }\n            }\n          } catch {\n            // sequence is disposed\n          }\n      });\n    }\n  }\n\n  /**\n   * Creates a new Sequence instance.\n   * @param abortSignal - Optional AbortSignal to cancel pending operations\n   */\n  constructor(abortSignal?: AbortSignal) {\n    super(abortSignal);\n    this.#queue = new RingBuffer();\n    this.#nextSignal = new Signal(abortSignal);\n    this.#sendSignal = new Signal(abortSignal);\n  }\n\n  /**\n   * Returns the number of values currently queued.\n   *\n   * @returns The current queue size\n   */\n  get size(): number {\n    return this.#queue.length;\n  }\n\n  /**\n   * Waits until the queue size drops to or below the specified capacity.\n   * Useful for implementing backpressure - producers can wait before adding more items.\n   *\n   * @param capacity The maximum queue size to wait for\n   * @returns A promise that resolves when the queue size is at or below capacity\n   *\n   * @example\n   * ```typescript\n   * // Producer with backpressure control\n   * const sequence = new Sequence<string>();\n   *\n   * // Wait if queue has more than 10 items\n   * await sequence.reserve(10);\n   * sequence.emit('new item'); // Safe to add, queue has space\n   * ```\n   */\n  async reserve(capacity: number): Promise<void> {\n    while (this.#queue.length > capacity) {\n      await this.#sendSignal;\n    }\n  }\n\n  /**\n   * Pushes a value onto the queue. Wakes any pending `receive()` waiter.\n   *\n   * @param value - The value to enqueue.\n   * @returns `true` if the sequence is still active.\n   */\n  emit(value: T): boolean {\n    const ok = !this.disposed;\n    if (ok) {\n      this.#queue.push(value);\n    }\n    this.#nextSignal.emit();\n    return ok;\n  }\n\n  /**\n   * Consumes and returns the next value from the queue.\n   * If the queue is empty, waits for a value to be added.\n   * Values are consumed in FIFO order.\n   * If the sequence has been aborted or disposed, this method rejects with Error('Disposed').\n   *\n   * @returns A promise that resolves with the next value\n   *\n   * @example\n   * ```typescript\n   * const sequence = new Sequence<number>();\n   *\n   * // Consumer waits for values\n   * const valuePromise = sequence.receive();\n   *\n   * // Producer adds value\n   * sequence.emit(42);\n   *\n   * // Consumer receives it\n   * const value = await valuePromise; // 42\n   * ```\n   */\n  async receive(): Promise<T> {\n    while (!this.#queue.length) {\n      await this.#nextSignal;\n    }\n    this.#sendSignal.emit();\n    return this.#queue.shift()!;\n  }\n\n  /**\n   * Disposes of the sequence, rejecting any pending `receive()` waiters.\n   * Called by `[Symbol.dispose]()` (inherited from Async) when using the `using` declaration.\n   */\n  dispose(): void {\n    this.#sendSignal[Symbol.dispose]();\n    this.#nextSignal[Symbol.dispose]();\n  }\n}\n"],"names":["Sequence","Async","Symbol","toStringTag","merge","target","sequences","source","queueMicrotask","disposed","sink","value","abortSignal","RingBuffer","Signal","size","length","reserve","capacity","emit","ok","push","receive","shift","dispose"],"mappings":";;;;+BAmCaA;;;eAAAA;;;0BAnCS;2BACC;+BACI;AAiCpB,MAAMA,iBAAoBC,eAAK;IACpC,CAAA,KAAM,CAAgB;IACtB,CAAA,UAAW,CAAe;IAC1B,CAAA,UAAW,CAAe;IAEjB,CAACC,OAAOC,WAAW,CAAC,GAAG,WAAW;IA6B3C,OAAOC,MAASC,MAAmB,EAAE,GAAGC,SAAwB,EAAQ;QACtE,KAAK,MAAMC,UAAUD,UAAW;YAC9BE,eAAe;gBACb,IAAI,CAACH,OAAOI,QAAQ,EAClB,IAAI;oBACF,MAAMC,OAAOL,OAAOK,IAAI;oBACxB,WAAW,MAAMC,SAASJ,OAAQ;wBAChC,IAAI,CAACG,KAAKC,QAAQ;4BAChB;wBACF;oBACF;gBACF,EAAE,OAAM,CAER;YACJ;QACF;IACF;IAMA,YAAYC,WAAyB,CAAE;QACrC,KAAK,CAACA;QACN,IAAI,CAAC,CAAA,KAAM,GAAG,IAAIC,yBAAU;QAC5B,IAAI,CAAC,CAAA,UAAW,GAAG,IAAIC,iBAAM,CAACF;QAC9B,IAAI,CAAC,CAAA,UAAW,GAAG,IAAIE,iBAAM,CAACF;IAChC;IAOA,IAAIG,OAAe;QACjB,OAAO,IAAI,CAAC,CAAA,KAAM,CAACC,MAAM;IAC3B;IAmBA,MAAMC,QAAQC,QAAgB,EAAiB;QAC7C,MAAO,IAAI,CAAC,CAAA,KAAM,CAACF,MAAM,GAAGE,SAAU;YACpC,MAAM,IAAI,CAAC,CAAA,UAAW;QACxB;IACF;IAQAC,KAAKR,KAAQ,EAAW;QACtB,MAAMS,KAAK,CAAC,IAAI,CAACX,QAAQ;QACzB,IAAIW,IAAI;YACN,IAAI,CAAC,CAAA,KAAM,CAACC,IAAI,CAACV;QACnB;QACA,IAAI,CAAC,CAAA,UAAW,CAACQ,IAAI;QACrB,OAAOC;IACT;IAwBA,MAAME,UAAsB;QAC1B,MAAO,CAAC,IAAI,CAAC,CAAA,KAAM,CAACN,MAAM,CAAE;YAC1B,MAAM,IAAI,CAAC,CAAA,UAAW;QACxB;QACA,IAAI,CAAC,CAAA,UAAW,CAACG,IAAI;QACrB,OAAO,IAAI,CAAC,CAAA,KAAM,CAACI,KAAK;IAC1B;IAMAC,UAAgB;QACd,IAAI,CAAC,CAAA,UAAW,CAACtB,OAAOsB,OAAO,CAAC;QAChC,IAAI,CAAC,CAAA,UAAW,CAACtB,OAAOsB,OAAO,CAAC;IAClC;AACF"}