Class AsyncEventQueue

packages/runtime/src/async-queue.ts:26–162

 */

export class AsyncEventQueue<T> implements AsyncIterable<T> {
  private buf: T[] = [];
  private waiters: Array<{
    resolve: (r: IteratorResult<T>) => void;
    reject: (e: Error) => void;
  }> = [];
  private closed = false;
  private err: Error | null = null;
  /** Monotonic count of events accepted by push(). */
  pushedCount = 0;
  /** Monotonic count of events the consumer has fully processed. */
  consumedCount = 0;
  /** Set when the consumer abandoned the stream; progress waiters must not block on it. */
  consumerDetached = false;
  private progressWaiters: Array<() => void> = [];

  push(item: T): void {
    if (!this.canEnqueue()) return;
    this.enqueue(item);
  }

  /**
   * Enqueue one item and resolve only after the consumer has fully processed
   * that exact sequence. Unlike `push()`, an enqueue rejected by queue state is
   * observable by the producer.
   */
  pushAndWaitUntilConsumed(item: T): Promise<void> {
    if (this.err) return Promise.reject(this.err);
    if (this.consumerDetached) return Promise.reject(consumerDetachedError());
    if (this.closed) return Promise.reject(queueClosedError());
    const sequence = this.enqueue(item);
    return this.waitUntilConsumed(sequence);
  }

  /**
   * Wait through the producer boundary captured at call time. Items enqueued
   * later do not extend the wait.
   */
  waitUntilConsumedThroughCurrent(): Promise<void> {
    return this.waitUntilConsumed(this.pushedCount);
  }

  private canEnqueue(): boolean {
    return !this.closed && !this.err && !this.consumerDetached;
  }

  private enqueue(item: T): number {
    const sequence = ++this.pushedCount;
    const w = this.waiters.shift();
    if (w) {
      w.resolve({ value: item, done: false });
    } else {
      this.buf.push(item);
    }
    this.wake();
    return sequence;
  }