Multi-Hoster-Upload/lib/semaphore.js

/**
 * FIFO Semaphore for per-hoster concurrency control.
 * acquire(signal?) blocks until a slot is available or the signal aborts.
 * release() frees a slot.
 */
class Semaphore {
  constructor(limit) {
    this.limit = Math.max(1, limit || 1);
    this.active = 0;
    this.queue = []; // { resolve, reject, signal?, onAbort? }
  }

  acquire(signal) {
    return new Promise((resolve, reject) => {
      if (signal && signal.aborted) {
        reject(new Error('Aborted'));
        return;
      }

      if (this.active < this.limit) {
        this.active++;
        resolve();
        return;
      }

      const entry = { resolve, reject };

      if (signal) {
        entry.signal = signal;
        entry.onAbort = () => {
          const idx = this.queue.indexOf(entry);
          if (idx !== -1) this.queue.splice(idx, 1);
          reject(new Error('Aborted'));
        };
        signal.addEventListener('abort', entry.onAbort, { once: true });
      }

      this.queue.push(entry);
    });
  }

  _cleanupEntry(entry) {
    if (entry.signal && entry.onAbort) {
      entry.signal.removeEventListener('abort', entry.onAbort);
    }
  }

  release() {
    if (this.queue.length > 0) {
      const entry = this.queue.shift();
      this._cleanupEntry(entry);
      entry.resolve();
    } else {
      this.active = Math.max(0, this.active - 1);
    }
  }

  updateLimit(newLimit) {
    this.limit = Math.max(1, newLimit || 1);
    while (this.active < this.limit && this.queue.length > 0) {
      this.active++;
      const entry = this.queue.shift();
      this._cleanupEntry(entry);
      entry.resolve();
    }
  }

  get pending() {
    return this.queue.length;
  }
}

module.exports = Semaphore;