Bounded Async Queue With Backpressure

export class BoundedQueue {
  constructor(capacity) {
    if (!Number.isInteger(capacity) || capacity < 1) {
      throw new RangeError("capacity must be a positive integer");
    }
    this.capacity = capacity;
    this.items = [];
    this.consumerWaiting = [];
    this.producerWaiting = [];
    this.closed = false;
  }

  get size() {
    return this.items.length;
  }

  async push(item) {
    if (this.closed) throw new Error("queue is closed");
    while (this.items.length >= this.capacity) {
      await new Promise((resolve) => this.producerWaiting.push(resolve));
      if (this.closed) throw new Error("queue is closed");
    }
    this.items.push(item);
    this.wakeConsumer();
  }

  async take() {
    while (this.items.length === 0) {
      if (this.closed) return undefined;
      await new Promise((resolve) => this.consumerWaiting.push(resolve));
    }
    const item = this.items.shift();
    this.wakeProducer();
    return item;
  }

  close() {
    this.closed = true;
    for (const resolve of this.consumerWaiting) resolve();
    for (const resolve of this.producerWaiting) resolve();
    this.consumerWaiting = [];
    this.producerWaiting = [];
  }

  async run(consumer) {
    const results = [];
    while (true) {
      const item = await this.take();
      if (item === undefined && this.closed) break;
      results.push(await consumer(item));
    }
    return results;
  }

  wakeConsumer() {
    const resolve = this.consumerWaiting.shift();
    if (resolve) resolve();
  }

  wakeProducer() {
    const resolve = this.producerWaiting.shift();
    if (resolve) resolve();
  }
}