change queueSizeLimit, buffSizeLimit to limit

Author: xsubject

README.md

@@ -24,8 +24,7 @@ The constructor takes a `QueueConfig` object as an argument.
 ### QueueConfig<T, R>
 
 - `worker` (optional): Worker function that processes tasks of type `T` and returns a Promise of type `R`.
-- `queueSizeLimit` (optional): Limits the maximum allowed size of the task queue.
-- `buffSizeLimit` (optional): Limits the maximum allowed size of the results buffer that have not yet been retrieved.
+- `limit` (optional): Limits the maximum allowed size of the tasks in queue.
 - `workPolicy`: Configures how tasks are processed in the queue.
   - `after-add` (default): The worker is started immediately after calling the `push()` method.
   - `async-cycle-one`: The worker is started in a separate asynchronous call with an optional `interval`.

src/Queue.spec.ts

@@ -2,7 +2,8 @@ import { describe, expect, test } from '@jest/globals'
 import { Queue } from './Queue'
 import { QueueConfig } from './QueueConfig'
 import { WorkerFn } from './WorkerFn'
-
+const wait = async (timeout: number) =>
+    new Promise((done) => setTimeout(done, timeout))
 describe('Queue', () => {
     let queue: Queue<any, any>
 
@@ -41,16 +42,31 @@ describe('Queue', () => {
         expect(result).toBe('Custom: 1')
     })
 
-    test('should block push when queue size limit reached', async () => {
-        // TODO
-    })
+    test('should block push when limit reached', async () => {
+        const tmp: any = {}
 
-    test('should block push when buffer size limit reached', async () => {
-        // TODO
-    })
+        const worker: WorkerFn<number, string> = async (item) => {
+            tmp[item] = true
+            return `Processed: ${item}`
+        }
+        queue = new Queue<number, string>({ worker, limit: 3 })
+
+        const p = queue.push([1, 2, 3, 4, 5])
+
+        await wait(10)
+
+        expect(queue.buff).toBe(3)
+        expect(tmp[1]).toBe(true)
+        expect(tmp[2]).toBe(true)
+        expect(tmp[3]).toBe(true)
+        expect(tmp[4]).toBe(undefined)
+        await queue.get()
+
+        await wait(10)
 
-    test('should push successfully after getting an item when limit reached', async () => {
-        // TODO
+        expect(queue.buff).toBe(3)
+        expect(tmp[4]).toBe(true)
+        expect(tmp[5]).toBe(undefined)
     })
 
     test('should work with after-add policy', async () => {

src/Queue.ts

@@ -4,8 +4,7 @@ import { Task } from './Task'
 import { WorkerFn } from './WorkerFn'
 import { Response } from './Respose'
 export class Queue<T, R> {
-    public readonly queueSizeLimit?: number
-    public readonly buffSizeLimit?: number
+    public readonly limit?: number
     public readonly workPolicy: string
 
     private readonly _worker?: WorkerFn<T, R>
@@ -33,11 +32,10 @@ export class Queue<T, R> {
             ...config,
         }
 
-        const { queueSizeLimit, buffSizeLimit, worker, workPolicy } = config
+        const { limit, worker, workPolicy } = config
 
         this.workPolicy = workPolicy
-        this.queueSizeLimit = queueSizeLimit
-        this.buffSizeLimit = buffSizeLimit
+        this.limit = limit
         this._worker = worker
 
         if (config.workPolicy === 'async-cycle-one') {
@@ -157,15 +155,9 @@ export class Queue<T, R> {
 
         const release = await this._mu.acquire()
 
-        if (this._musl.isLocked()) {
-            await this._musl.waitForUnlock()
-        }
+        await this._musl.waitForUnlock()
 
-        if (
-            (this.queueSizeLimit &&
-                this._queue.length >= this.queueSizeLimit) ||
-            (this.buffSizeLimit && this._buff.length >= this.buffSizeLimit)
-        ) {
+        if (this.limit && this.length >= this.limit) {
             await this._musl.acquire()
             await this._musl.waitForUnlock()
         }

src/QueueConfig.ts

@@ -2,8 +2,7 @@ import { WorkerFn } from './WorkerFn'
 
 export type QueueConfig<T, R> = {
     worker?: WorkerFn<T, R>
-    queueSizeLimit?: number
-    buffSizeLimit?: number
+    limit?: number
 } & (
     | {
           workPolicy: 'after-add'