Asynchronous Generic Resource Pooler

Question

A personal project I'm working on needs Worker (thread) pool. The libraries I looked at were too high-level for my needs, as I simply needed access to the Worker itself while the libraries offered task-runner abstractions that were difficult to work around.

I decided to write my own, but I figured I would make it generic so that it would work with not only workers, but any arbitrary resource.

interface ResourceFactory<T, U = T> {
  create(): PromiseLike<T>;
  dispose?(resource: T): PromiseLike<T>;
  access?(resource: T): PromiseLike<U>;
}


class AsyncQueue<T> {
  #elements: T[] = [];
  #waiting: ((el: T) => void)[] = [];
  get size() { return this.#elements.length }

  enqueue(el: T) {
    const next = this.#waiting.shift();
    if (next) {
      next(el);
    } else {
      this.#elements.push(el);
    }
  }
  async dequeue() {
    const next = this.#elements.shift();
    if (next) {
      return next;
    } else {
      const defer = new Promise<T>(res => this.#waiting.push(res))
      return defer;
    }
  }
}

class ResourcePooler<T, U> {
  factory: ResourceFactory<T, U>;
  targetSize: number = 0;
  currentSize: number = 0;
  resourceQueue: AsyncQueue<T> = new AsyncQueue();

  constructor({
    factory,
  }: { factory: ResourceFactory<T, U> }) {
    this.factory = factory;
  }

  async use<O>(task: (resource: U) => Promise<O>, waitForDispose = false): Promise<O> {
    const resource = await this.resourceQueue.dequeue();
    const accessor = this.factory.access ?? (resource => resource);
    const accessed = await accessor(resource) as U;
    const result = await task(accessed);
    
    if (this.currentSize > this.targetSize) {
      // Get rid of excess resources
      const diposing = this.factory.dispose?.(resource);
      if (waitForDispose) await diposing;
      this.currentSize--;
    } else {
      this.resourceQueue.enqueue(resource);
    }

    return result;
  }

  private resizing = false;

  async resize(newSize: number) {
    // prevent conflicting resizes
    if (this.resizing) return;
    this.resizing = true;

    this.targetSize = newSize;

    while (newSize > this.currentSize) {
      this.currentSize++;
      const next = await this.factory.create();
      this.resourceQueue.enqueue(next);
    }

    
    while (newSize < this.currentSize) {
      this.use(async () => {}, true)
    }
  }
}

async function createPool<T, U = T>(factory: ResourceFactory<T, U>, size: number = 8) {
  const pooler = new ResourcePooler<T, U>({
    factory,
  })
  pooler.resize(size);
  return pooler;
}

I wrote some consumer code to test it:

let counter = 0;

class Foo {
  id: number;
  constructor() {
    this.id = ++counter;
  }
  getReady() {
    return new Promise<void>(res => {
      setTimeout(() =>{ 
        res()
        }, 1);
    })
  }
}

async function main() {
   const pooler = await createPool({
    create: async () => new Foo()
   });
   for (let x = 0; x < 200; x++) {
    pooler.use(foo => new Promise<number>(res => {
      console.log("Doing some work with", foo.id);
      setTimeout(() => res(foo.id), 1000*Math.random());
    })).then(result => console.log(result))
   }
}

main();

It seems to behave correctly. I wonder if I could implement any of this better; in particular the pool resizing.