tests: promise and priority queue tests

Author: r4mmer

__tests__/models/priority_queue.test.ts

@@ -0,0 +1,47 @@
+/**
+ * Copyright (c) Hathor Labs and its affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+import PriorityQueue from '../../src/models/priority_queue';
+
+test('PriorityQueue operations', () => {
+  const q = new PriorityQueue<string>();
+  expect(q.size).toEqual(0);
+  // method: makeNode
+  expect(PriorityQueue.makeNode(123, 'foobar')).toMatchObject({
+    priority: 123,
+    value: 'foobar',
+  });
+
+  // methods: push and pop
+  q.push(PriorityQueue.makeNode(10, 'foobar1'));
+  expect(q.size).toEqual(1);
+  expect(q.isEmpty()).toBeFalsy();
+  expect(q.pop()).toEqual('foobar1');
+  expect(q.isEmpty()).toBeTruthy();
+
+  // method: add
+  const elements = [
+    { priority: 10, value: 'foo' },
+    { priority: 1,  value: 'baz' },
+    { priority: 5,  value: 'bar' },
+  ].map(el => PriorityQueue.makeNode(el.priority, el.value));
+  q.add(...elements);
+  expect(q.size).toEqual(3);
+  expect(q.isEmpty()).toBeFalsy();
+  expect(q.pop()).toEqual('foo');
+  expect(q.size).toEqual(2);
+  expect(q.isEmpty()).toBeFalsy();
+  expect(q.peek()).toEqual('bar');
+  expect(q.pop()).toEqual('bar');
+  expect(q.size).toEqual(1);
+  expect(q.isEmpty()).toBeFalsy();
+  expect(q.peek()).toEqual('baz');
+  expect(q.pop()).toEqual('baz');
+  expect(q.size).toEqual(0);
+  expect(q.isEmpty()).toBeTruthy();
+});
+

__tests__/models/promise_queue.test.ts

@@ -0,0 +1,55 @@
+/**
+ * Copyright (c) Hathor Labs and its affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+import PromiseQueue from '../../src/models/promise_queue';
+
+describe('PromiseQueue operations', () => {
+  it('should start unpaused and be pausable', () => {
+    const q = new PromiseQueue();
+    expect(q.isPaused).toBeFalsy();
+    q.stop();
+    expect(q.isPaused).toBeTruthy();
+    q.continue();
+    expect(q.isPaused).toBeFalsy();
+  });
+
+  it('should be able to set the concurrency', async () => {
+    const q = new PromiseQueue();
+    expect(q.concurrent).toEqual(1);
+    q.concurrent = 3;
+    expect(q.concurrent).toEqual(3);
+
+    // Check that the concurrent actually runs only 2 tasks
+    q.concurrent = 2;
+
+    const task = async () => {
+      return new Promise<void>(resolve => {
+        setTimeout(resolve, 2000);
+      });
+    };
+    q.add(task);
+    q.add(task);
+    q.add(task);
+    // wait 50ms so jobs can start
+    await new Promise<void>(resolve => {
+      setTimeout(resolve, 50);
+    });
+    // Expect that only 2 jobs are running
+    expect(q.jobsRunning).toEqual(2);
+
+    // Wait for either all tasks to finish or a timeout.
+    await Promise.race([
+      new Promise<void>(resolve => {
+        // idle is sent when the queue  is empty and there are no events running
+        q.on('idle', resolve);
+      }),
+      new Promise<void>((_, reject) => {
+        setTimeout(reject, 5000);
+      }),
+    ]);
+  });
+});

src/models/priority_queue.ts

@@ -1,3 +1,9 @@
+/**
+ * Copyright (c) Hathor Labs and its affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
 /* eslint class-methods-use-this: ["error", { "exceptMethods": ["parent", "left", "right", "comparator"] }] */
 
 interface PriorityQueueNode<T> {

src/models/promise_queue.ts

@@ -1,3 +1,9 @@
+/**
+ * Copyright (c) Hathor Labs and its affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
 import EventEmitter from 'events';
 import PriorityQueue from './priority_queue';
 
@@ -22,7 +28,7 @@ export default class PromiseQueue extends EventEmitter {
 
   #isPaused = false;
 
-  #intervalId: ReturnType<typeof setInterval> | null = null;
+  #intervalId: ReturnType<typeof setTimeout> | null = null;
 
   constructor() {
     super();
@@ -32,9 +38,13 @@ export default class PromiseQueue extends EventEmitter {
   }
 
   #startInterval() {
-    this.#clearInterval();
-    this.#intervalId = setInterval(() => {
-      this.processQueue();
+    this.processQueue();
+    this.#intervalId = setTimeout(() => {
+      if (this.#isPaused) {
+        this.#clearInterval();
+        return;
+      }
+      this.#startInterval();
     }, 1000);
   }
 
@@ -45,21 +55,24 @@ export default class PromiseQueue extends EventEmitter {
     }
   }
 
-  get isPaused() {
+  public get isPaused() {
     return this.#isPaused;
   }
 
-  stop() {
+  public stop() {
     this.#isPaused = true;
     this.#clearInterval();
   }
 
-  continue() {
+  public continue() {
     this.#isPaused = false;
-    this.processQueue();
     this.#startInterval();
   }
 
+  public get jobsRunning() {
+    return this.#jobsRunning;
+  }
+
   #next() {
     this.emit('next');
     this.#jobsRunning--;
@@ -70,18 +83,21 @@ export default class PromiseQueue extends EventEmitter {
     return this.#jobsRunning < this.concurrent;
   }
 
-  get concurrent() {
+  public get concurrent() {
     return this.#allowedConcurrentJobs;
   }
 
-  set concurrent(value) {
+  public set concurrent(value) {
     if (value < 1) {
       throw new Error('Cannot have less than 1 job running.');
     }
     this.#allowedConcurrentJobs = value;
   }
 
   #tryToStartNextJob(): boolean {
+    if (this.#isPaused) {
+      return false;
+    }
     if (this.#queue.isEmpty()) {
       // No more tasks to run
       this.emit('queue_empty');
@@ -153,6 +169,8 @@ export default class PromiseQueue extends EventEmitter {
           }
         })
       );
+      // Try to start the job we enqueued and any other we can start
+      this.processQueue();
     });
   }
 }