✨ Add TaskQueue

Author: Tomas2D

README.md

@@ -18,6 +18,7 @@ Does this library help you? Please give it a ⭐️!
 - Auto rejecting promises removed from the shared data structure (`TaskMap`, which is an extension of a `Map` data structure)
 - Zero dependencies
 - Map with limited space (sliding-window)
+- Task Queue mechanism
 
 ## 🚀 Installation
 
@@ -110,16 +111,56 @@ app.get('/observations/:date', async function (req, res) {
 })
 ```
 
+**Task Queue**
+
+```typescript
+// Type Definition
+declare class TaskQueue<T> {
+  constructor(queueSize?: number); // default is 1
+  clear(): Promise<void>;
+  execute(fn: () => Promise<T>): Promise<T>;
+  getStats(): {
+    processingCount: number;
+    waitingCount: number;
+  };
+}
+```
+
+You want to use the Task Queue when you want to throttle requests/executions of given part of your system.
+
+```typescript
+const queue = new TaskQueue<IScrapeResult>(5)
+
+app.get('/scrape/:source', async function (req, res) {
+  const source = req.params.source
+  const response = await queue.execute(async () => {
+    // this function will run by the queue mechanism
+    return service.scrape(source)
+  })
+  
+  res.json(response)
+})
+
+app.get('/scrape/status', async function (req, res) {
+  const { processingCount, waitingCount } = queue.getStats()
+  
+  return res.json({
+    processingCount,
+    waitingCount,
+  })
+})
+```
+
 **Sliding window**
 ```typescript
 // Type Definition
 declare class SlidingTaskMap<K, V extends Deletable> extends TaskMap<K, V> {
-    constructor(windowSize: number, ttl?: number);
-    set(key: K, value: V, customTTL?: number): this; // create/override 
-    delete(key: K): boolean; // remove single entry
-    clear(): void;  // removes all entries
-    pop(): boolean; // removes oldest entry, false for an empty window
-    shift(): boolean;  // removes newest entry, false for an empty window
+  constructor(windowSize: number, ttl?: number);
+  set(key: K, value: V, customTTL?: number): this; // create/override 
+  delete(key: K): boolean; // remove single entry
+  clear(): void;  // removes all entries
+  pop(): boolean; // removes oldest entry, false for an empty window
+  shift(): boolean;  // removes newest entry, false for an empty window
 }
 ```
 
@@ -133,6 +174,31 @@ const WINDOW_SIZE = 10
 const TTL = 60 * 1000 // data will persist 60 seconds in a cache
 const tasks = new SlidingTaskMap<string, number[]>(WINDOW_SIZE, TTL)
 
+app.get('/calculation/:date', async function () {
+  const date = req.params.date
+
+  if (!tasks.has(date)) {
+    const task = new Task<void>()
+    tasks.set(date, task)
+
+    const data = await fetchData(date)
+    task.resolve(data)
+  }
+
+  return tasks.get(date)
+})
+```
+
+When your map size reaches a specified threshold, the oldest values will be
+removed. You can be then sure that the size of the map will never overflow your memory.
+
+```typescript
+import { Task, SlidingTaskMap } from 'promise-based-task'
+
+const WINDOW_SIZE = 10
+const TTL = 60 * 1000 // data will persist 60 seconds in a cache
+const tasks = new SlidingTaskMap<string, number[]>(WINDOW_SIZE, TTL)
+
 app.get('/calculation/:date', async function () {
   const date = req.params.date
 

src/TaskQueue.ts

@@ -0,0 +1,79 @@
+import { Task } from './Task'
+import { TaskDestroyedException } from './error';
+
+interface IWaitingTask<T> {
+  task: Task<T>
+  resolver: () => Promise<void>
+}
+
+export class TaskQueue<T> {
+  private _waitingQueue: Array<IWaitingTask<T>> = []
+  private _unlockQueue: Array<Task<void>> = []
+
+
+  private readonly _queueSize: number
+  private _processingCount = 0
+
+  constructor(queueSize?: number) {
+    this._queueSize = Math.max(Number(queueSize), 1)
+  }
+
+  private async _internalSync() {
+    if (this._processingCount >= this._queueSize) {
+      const waitForUnlockTask = new Task<void>()
+      this._unlockQueue.push(waitForUnlockTask)
+      await waitForUnlockTask
+    }
+
+    const task = this._waitingQueue.shift()
+    if (task) {
+      await task.resolver()
+    }
+  }
+
+  async clear() {
+    this._waitingQueue.forEach(item => item.task.reject(new TaskDestroyedException()))
+    this._unlockQueue.forEach(item => item.reject(new TaskDestroyedException()))
+
+    this._unlockQueue.length = 0
+    this._waitingQueue.length = 0
+    this._processingCount = 0
+  }
+
+  async execute(fn: () => Promise<T>): Promise<T> {
+    const task = new Task<T>()
+    this._waitingQueue.push({
+      task,
+      resolver: async () => {
+        this._processingCount++
+
+        const cleanup = () => {
+          this._processingCount--
+          const unlockTask = this._unlockQueue.shift()
+          unlockTask?.resolve()
+        }
+
+        return fn()
+          .then((res) => {
+            cleanup();
+            task.resolve(res);
+          })
+          .catch(e => {
+            cleanup();
+            task.reject(e);
+          })
+      }
+    })
+
+    this._internalSync().catch(() => {})
+
+    return task
+  }
+
+  getStats() {
+    return {
+      processingCount: this._processingCount,
+      waitingCount: this._waitingQueue.length,
+    }
+  }
+}

src/index.ts

@@ -1,5 +1,6 @@
 export * from './Task';
 export * from './TaskMap';
 export * from './SlidingTaskMap';
+export * from './TaskQueue';
 export * from './types';
 export * from './error';

test/TaskQueue.test.ts

@@ -0,0 +1,99 @@
+import { TaskQueue } from '../src';
+
+describe('TaskQueue', () => {
+  const sleepFactory = <T>(ms: number, res?: T) => {
+    return () => new Promise<T>((resolve) => setTimeout(() => resolve(res), ms));
+  };
+
+  beforeEach(() => {
+    jest.useRealTimers();
+  });
+
+  it('Can process single item', async () => {
+    const queue = new TaskQueue<number>(1);
+    expect(queue.getStats().processingCount).toBe(0);
+    expect(queue.getStats().waitingCount).toBe(0);
+
+    const result = queue.execute(sleepFactory(100, 42));
+    expect(queue.getStats().processingCount).toBe(1);
+    expect(queue.getStats().waitingCount).toBe(0);
+    await expect(result).resolves.toBe(42);
+    expect(queue.getStats().processingCount).toBe(0);
+    expect(queue.getStats().waitingCount).toBe(0);
+  });
+
+  it('Can process multiple items', async () => {
+    const queue = new TaskQueue<number>(3);
+    const results = Promise.all([
+      queue.execute(sleepFactory(100, 1)),
+      queue.execute(sleepFactory(100, 2)),
+      queue.execute(sleepFactory(100, 3)),
+      queue.execute(sleepFactory(100, 4)),
+      queue.execute(sleepFactory(100, 5)),
+    ]);
+    const preStats = queue.getStats();
+    expect(preStats.processingCount).toBe(3);
+    expect(preStats.waitingCount).toBe(2);
+
+    await expect(results).resolves.toStrictEqual([1, 2, 3, 4, 5]);
+
+    const postStats = queue.getStats();
+    expect(postStats.processingCount).toBe(0);
+    expect(postStats.waitingCount).toBe(0);
+  });
+
+  it('Can process multiple items', async () => {
+    const queue = new TaskQueue<number>(3);
+    const results = Promise.all([
+      queue.execute(sleepFactory(50, 1)),
+      queue.execute(sleepFactory(50, 2)),
+      queue.execute(sleepFactory(50, 3)),
+      queue.execute(sleepFactory(50, 4)),
+      queue.execute(sleepFactory(50, 5)),
+    ]);
+    const preStats = queue.getStats();
+    expect(preStats.processingCount).toBe(3);
+    expect(preStats.waitingCount).toBe(2);
+
+    await expect(results).resolves.toStrictEqual([1, 2, 3, 4, 5]);
+
+    const postStats = queue.getStats();
+    expect(postStats.processingCount).toBe(0);
+    expect(postStats.waitingCount).toBe(0);
+  });
+
+  it('Can clear/destroy the queue', async () => {
+    const queue = new TaskQueue<number>(3);
+    const results = Promise.allSettled([
+      queue.execute(sleepFactory(50, 1)),
+      queue.execute(sleepFactory(50, 2)),
+      queue.execute(sleepFactory(50, 3)),
+      queue.execute(sleepFactory(50, 4)),
+      queue.execute(sleepFactory(50, 5)),
+    ]).then((results) => results.map((result) => result.status));
+
+    await sleepFactory(0)();
+    await queue.clear();
+
+    await expect(results).resolves.toMatchInlineSnapshot(`
+            Array [
+              "fulfilled",
+              "fulfilled",
+              "fulfilled",
+              "rejected",
+              "rejected",
+            ]
+          `);
+  });
+
+
+  it('Will resolve in correct order', async () => {
+    const queue = new TaskQueue<number>(1);
+    const result = Promise.race([
+      queue.execute(sleepFactory(500, 1)),
+      queue.execute(sleepFactory(50, 2)),
+    ])
+
+    await expect(result).resolves.toBe(1)
+  });
+});