Merge pull request #1 from prodialabs/queue-depths-functions

`*`: add `getQueueDepths()` and `getRetryDepths()`

.github/workflows/test.yml

@@ -2,8 +2,7 @@ name: Run Tests
 
 on:
   push:
-    branches:
-      - master
+
 
 jobs:
   test:

main.ts

@@ -213,11 +213,51 @@ export const createMultiQueue: CreateMultiQueue = <Queue, Job>(
 		return typeof job === "string" ? JSON.parse(job) as Job : undefined;
 	};
 
+	const getQueueDepths = async () => {
+		const queues = await redis.zrangebyscore(
+			queueDepthKey,
+			"0",
+			"+inf",
+		);
+
+		const depths = new Map<Queue, number>();
+
+		for (const queue of queues) {
+			depths.set(
+				JSON.parse(queue) as Queue,
+				await redis.zcard(`${queueKeyPrefix}${queue}`),
+			);
+		}
+
+		return depths;
+	};
+
+	const getRetryDepths = async () => {
+		const queues = await redis.zrangebyscore(
+			queueDepthKey,
+			"0",
+			"+inf",
+		);
+
+		const depths = new Map<Queue, number>();
+
+		for (const queue of queues) {
+			depths.set(
+				JSON.parse(queue) as Queue,
+				await redis.zcard(`${retryKeyPrefix}${queue}`),
+			);
+		}
+
+		return depths;
+	};
+
 	return {
 		push,
 		pop,
 		complete,
 		getDeepest,
 		popAny,
+		getQueueDepths,
+		getRetryDepths,
 	};
 };

test.ts

@@ -37,17 +37,67 @@ await (async () => {
 	await mq.push(myQueue, secondJob, 500);
 	await mq.push(myQueue, firstJob, 250);
 
+	assertEquals(
+		await mq.getQueueDepths(),
+		new Map([
+			[myQueue, 2],
+		]),
+		"getQueueDepths() must return correct queue sizes",
+	);
+
+	assertEquals(
+		await mq.getRetryDepths(),
+		new Map([
+			[myQueue, 0],
+		]),
+		"getRetryDepths() must return correct queue sizes",
+	);
+
+	// pop one element
 	assertEquals(
 		await mq.pop(myQueue),
 		firstJob,
 		"pop() must return correct job",
 	);
 
+	assertEquals(
+		await mq.getQueueDepths(),
+		new Map([
+			[myQueue, 1],
+		]),
+		"getQueueDepths() must return correct queue sizes",
+	);
+
+	assertEquals(
+		await mq.getRetryDepths(),
+		new Map([
+			[myQueue, 1],
+		]),
+		"getRetryDepths() must return correct queue sizes",
+	);
+
+	// pop another element
 	assertEquals(
 		await mq.pop(myQueue),
 		secondJob,
 		"pop() must return correct job",
 	);
+
+	assertEquals(
+		await mq.getQueueDepths(),
+		new Map([
+			[myQueue, 0],
+		]),
+		"getQueueDepths() must return correct queue sizes",
+	);
+
+	assertEquals(
+		await mq.getRetryDepths(),
+		new Map([
+			[myQueue, 2],
+		]),
+		"getRetryDepths() must return correct queue sizes",
+	);
 })();
 
 await (async () => {

types.ts

@@ -14,6 +14,8 @@ export type MultiQueue<Queue extends JsonValue, Job extends JsonValue> = {
 	complete: (queue: Queue, job: Job) => Promise<void>;
 	getDeepest: () => Promise<Queue | undefined>;
 	popAny: (queues?: Queue[]) => Promise<Job | undefined>;
+	getQueueDepths: () => Promise<Map<Queue, number>>;
+	getRetryDepths: () => Promise<Map<Queue, number>>;
 };
 
 export type CreateMultiQueueOptions = {