export default class ConsumerSetQueue<T> {
  private queue: Set<T>;
  private consumers: Array<Consumer<T>>;
  private closed: boolean = false;
  constructor() {
    this.queue = new Set();
    this.consumers = [];
  }
  push(value: T): void {
    if (!this.queue.has(value)) {
      // avoid duplicate entries in the queue
      this.queue.add(value);
      setTimeout(() => {
        this.checkQueue();
      }, 5); // make sure checkqueue is executed asynchronously and we don't block the calling processing
    }
  }
  addConsumer(consumer: Consumer<T>) {
    this.consumers.push(consumer);
    this.checkQueue();
  }
  async checkQueue() {
    const consumer = this.consumers.pop();
    if (!consumer) {
      return;
    }

    try {
      while (this.queue.size > 0) {
        const value: T = this.queue.keys().next().value;
        this.queue.delete(value);
        await consumer.consume(value);
      }
    } finally {
      if (this.closed) {
        consumer.close();
      } else {
        this.consumers.push(consumer);
      }
    }
  }

  close() {
    this.closed = true;
    while (this.consumers.length > 0) {
      const consumer = this.consumers.pop();
      consumer?.close();
    }
  }
}
export interface Consumer<T> {
  consume(value: T): Promise<void>;
  close(): void;
}