Job Queue in TypeScript

job-queue-example.ts

import { JobQueue } from "./job-queue";

export interface Device {
  send(command: string): Promise<void>;
}

const sleep = (ms: number) => new Promise<void>((fn) => setTimeout(fn, ms));

export class AsyncDevice implements Device {
  async send(command: string): Promise<void> {
    console.log("start", command);
    await sleep(1000);
    console.log("end", command);
  }
}

export class SyncDevice implements Device {
  private _device: Device;
  private _jobs: JobQueue;

  constructor(device: Device) {
    this._device = device;
    this._jobs = new JobQueue();
  }

  send(command: string): Promise<void> {
    return this._jobs.run(() => this._device.send(command));
  }
}

async function example(title: string, device: Device) {
  console.log(title);

  await Promise.all([
    // send two messages at once
    // the asynchronous device will run in parallel
    // the synchronous device will run in series
    device.send("one"),
    device.send("two"),
  ]);

  console.log();
}

async function main(): Promise<void> {
  const asyncDevice = new AsyncDevice();
  const syncDevice = new SyncDevice(asyncDevice);

  await example("AsyncDevice", asyncDevice);
  await example("SyncDevice", syncDevice);
}

main().catch(console.error);

// AsyncDevice
// start one
// start two
// end one
// end two

// SyncDevice
// start one
// end one
// start two
// end two

job-queue.ts

export interface Job<T = any> {
  run(): Promise<T>;
  resolve(value: T): void;
  reject(err: Error): void;
}

export class JobQueue {
  private _queue: Job[];
  private _wake: () => void;

  constructor() {
    this._queue = [];
    this._wake = Boolean;
    this._worker().catch(console.error);
  }

  async run<T>(run: () => Promise<T>): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      this._queue.push({ run, resolve, reject });
      this._wake();
    });
  }

  private async _worker(): Promise<void> {
    for (;;) {
      for (;;) {
        const job = this._queue.shift();

        if (typeof job === "undefined") {
          break;
        }

        await job.run().then(job.resolve, job.reject);
      }

      await new Promise<void>((wake) => {
        this._wake = wake;
      });
    }
  }
}