refactor: remove dependency on p-queue (#250)

Author: JustinBeckwith

package-lock.json

@@ -15,7 +15,6 @@
     "compile": "tsc -p .",
     "test": "c8 mocha build/test",
     "fix": "gts fix",
-    "codecov": "c8 report --reporter=json && codecov -f coverage/*.json",
     "lint": "gts lint",
     "build-binaries": "pkg . --out-path build/binaries",
     "docs-test": "npm link && linkinator ./README.md"
@@ -28,7 +27,6 @@
     "jsonexport": "^3.0.0",
     "marked": "^1.2.5",
     "meow": "^9.0.0",
-    "p-queue": "^6.2.1",
     "serve-handler": "^6.1.3",
     "server-destroy": "^1.0.1",
     "update-notifier": "^5.0.0"

package.json

@@ -1,52 +1,75 @@
-import PQueue from 'p-queue';
+import {EventEmitter} from 'events';
 
 export interface QueueOptions {
-  concurrency?: number;
+  concurrency: number;
 }
 
 export interface QueueItemOptions {
   delay?: number;
 }
 
+interface QueueItem {
+  fn: AsyncFunction;
+  timeToRun: number;
+}
+
+export declare interface Queue {
+  on(event: 'done', listener: () => void): this;
+}
+
 export type AsyncFunction = () => Promise<void>;
 
-export class Queue {
-  private q: PQueue;
-  private activeTimers = 0;
+export class Queue extends EventEmitter {
+  private q: Array<QueueItem> = [];
+  private activeFunctions = 0;
+  private concurrency: number;
 
   constructor(options: QueueOptions) {
-    this.q = new PQueue({
-      concurrency: options.concurrency,
-    });
+    super();
+    this.concurrency = options.concurrency;
   }
 
   add(fn: AsyncFunction, options?: QueueItemOptions) {
-    if (options?.delay) {
-      setTimeout(() => {
-        this.q.add(fn);
-        this.activeTimers--;
-      }, options.delay);
-      this.activeTimers++;
-    } else {
-      this.q.add(fn);
-    }
+    const delay = options?.delay || 0;
+    const timeToRun = Date.now() + delay;
+    this.q.push({
+      fn,
+      timeToRun,
+    });
+    setTimeout(() => this.tick(), delay);
   }
 
-  async onIdle() {
-    await this.q.onIdle();
-    await new Promise<void>(resolve => {
-      if (this.activeTimers === 0) {
-        resolve();
+  private tick() {
+    // Check if we're complete
+    if (this.activeFunctions === 0 && this.q.length === 0) {
+      this.emit('done');
+      return;
+    }
+
+    for (let i = 0; i < this.q.length; i++) {
+      // Check if we have too many concurrent functions executing
+      if (this.activeFunctions >= this.concurrency) {
         return;
       }
-      const timer = setInterval(async () => {
-        if (this.activeTimers === 0) {
-          await this.q.onIdle();
-          clearInterval(timer);
-          resolve();
-          return;
-        }
-      }, 500);
+      // grab the element at the front of the array
+      const item = this.q.shift()!;
+      // make sure this element is ready to execute - if not, to the back of the stack
+      if (item.timeToRun > Date.now()) {
+        this.q.push(item);
+      } else {
+        // this function is ready to go!
+        this.activeFunctions++;
+        item.fn().finally(() => {
+          this.activeFunctions--;
+          this.tick();
+        });
+      }
+    }
+  }
+
+  async onIdle() {
+    return new Promise<void>(resolve => {
+      this.on('done', () => resolve());
     });
   }
 }

src/queue.ts

@@ -37,11 +37,14 @@ describe('retries', () => {
 
     const {promise, resolve} = invertedPromise();
     const checker = new LinkChecker().on('retry', resolve);
-    const clock = sinon.useFakeTimers();
+    const clock = sinon.useFakeTimers({
+      shouldAdvanceTime: true,
+    });
     const checkPromise = checker.check({
       path: 'test/fixtures/basic',
       retry: true,
     });
+
     await promise;
     await clock.tickAsync(10_000);
     const results = await checkPromise;
@@ -62,7 +65,9 @@ describe('retries', () => {
 
     const {promise, resolve} = invertedPromise();
     const checker = new LinkChecker().on('retry', resolve);
-    const clock = sinon.useFakeTimers();
+    const clock = sinon.useFakeTimers({
+      shouldAdvanceTime: true,
+    });
     const checkPromise = checker.check({
       path: 'test/fixtures/basic',
       retry: true,
@@ -85,7 +90,9 @@ describe('retries', () => {
 
     const {promise, resolve} = invertedPromise();
     const checker = new LinkChecker().on('retry', resolve);
-    const clock = sinon.useFakeTimers();
+    const clock = sinon.useFakeTimers({
+      shouldAdvanceTime: true,
+    });
     const checkPromise = checker.check({
       path: 'test/fixtures/basic',
       retry: true,
@@ -125,7 +132,9 @@ describe('retries', () => {
 
     const {promise, resolve} = invertedPromise();
     const checker = new LinkChecker().on('retry', resolve);
-    const clock = sinon.useFakeTimers();
+    const clock = sinon.useFakeTimers({
+      shouldAdvanceTime: true,
+    });
     const checkPromise = checker.check({
       path: 'test/fixtures/retry',
       recurse: true,
@@ -177,7 +186,9 @@ describe('retries', () => {
         r2();
       }
     });
-    const clock = sinon.useFakeTimers();
+    const clock = sinon.useFakeTimers({
+      shouldAdvanceTime: true,
+    });
     const checkPromise = checker.check({
       path: 'test/fixtures/retry',
       recurse: true,