Feature/invoke event

source/index.ts

@@ -8,7 +8,7 @@ type Task<TaskResultType> =
 	| ((options: TaskOptions) => PromiseLike<TaskResultType>)
 	| ((options: TaskOptions) => TaskResultType);
 
-type EventName = 'active' | 'idle' | 'empty' | 'add' | 'next' | 'completed' | 'error';
+type EventName = 'active' | 'idle' | 'empty' | 'add' | 'next' | 'completed' | 'error' | 'invoke';
 
 /**
 Promise queue with concurrency control.
@@ -43,6 +43,9 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 
 	readonly #throwOnTimeout: boolean;
 
+	/** Use to assign a unique identifier to a promise function, if not explicitly specified */
+	#idAssigner = 1;
+
 	/**
 	Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
 
@@ -228,12 +231,21 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 		});
 	}
 
+	setPriority(id: string, priority: number) {
+		this.#queue.setPriority(id, priority);
+	}
+
 	/**
 	Adds a sync or async task to the queue. Always returns a promise.
 	*/
 	async add<TaskResultType>(function_: Task<TaskResultType>, options: {throwOnTimeout: true} & Exclude<EnqueueOptionsType, 'throwOnTimeout'>): Promise<TaskResultType>;
 	async add<TaskResultType>(function_: Task<TaskResultType>, options?: Partial<EnqueueOptionsType>): Promise<TaskResultType | void>;
 	async add<TaskResultType>(function_: Task<TaskResultType>, options: Partial<EnqueueOptionsType> = {}): Promise<TaskResultType | void> {
+		// Incase id is not defined
+		if (options.id === undefined) {
+			options.id = (this.#idAssigner++).toString();
+		}
+
 		options = {
 			timeout: this.timeout,
 			throwOnTimeout: this.#throwOnTimeout,
@@ -258,6 +270,7 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 						operation = Promise.race([operation, this.#throwOnAbort(options.signal)]);
 					}
 
+					this.emit('invoke', {id: options.id});
 					const result = await operation;
 					resolve(result);
 					this.emit('completed', result);

source/options.ts

@@ -69,6 +69,7 @@ export type QueueAddOptions = {
 	@default 0
 	*/
 	readonly priority?: number;
+	id?: string;
 } & TaskOptions & TimeoutOptions;
 
 export type TaskOptions = {

source/priority-queue.ts

@@ -32,6 +32,27 @@ export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOp
 		this.#queue.splice(index, 0, element);
 	}
 
+	setPriority(id: string, priority?: number) {
+		const existingIndex: number = this.#queue.findIndex((element: Readonly<PriorityQueueOptions>) => element.id === id);
+		const [item] = this.#queue.splice(existingIndex, 1);
+		if (item === undefined) {
+			return;
+		}
+
+		item.priority = priority ?? ((item.priority ?? 0) + 1);
+		if (this.size && this.#queue[this.size - 1]!.priority! >= priority!) {
+			this.#queue.push(item);
+			return;
+		}
+
+		const index = lowerBound(
+			this.#queue, item,
+			(a: Readonly<PriorityQueueOptions>, b: Readonly<PriorityQueueOptions>) => b.priority! - a.priority!,
+		);
+
+		this.#queue.splice(index, 0, item);
+	}
+
 	dequeue(): RunFunction | undefined {
 		const item = this.#queue.shift();
 		return item?.run;

source/queue.ts

@@ -5,4 +5,5 @@ export type Queue<Element, Options> = {
 	filter: (options: Readonly<Partial<Options>>) => Element[];
 	dequeue: () => Element | undefined;
 	enqueue: (run: Element, options?: Partial<Options>) => void;
+	setPriority: (id: string, priority: number) => void;
 };

test/test.ts

@@ -6,7 +6,8 @@ import inRange from 'in-range';
 import timeSpan from 'time-span';
 import randomInt from 'random-int';
 import pDefer from 'p-defer';
-import PQueue, {AbortError} from '../source/index.js';
+import PQueue from '../source/index.js';
+import type {QueueAddOptions} from '../source/index.js';
 
 const fixture = Symbol('fixture');
 
@@ -1134,3 +1135,41 @@ test('aborting multiple jobs at the same time', async t => {
 	await t.throwsAsync(task2, {instanceOf: DOMException});
 	t.like(queue, {size: 0, pending: 0});
 });
+
+test('.setPriority() - execute a promise before planned', async t => {
+	const result: string[] = [];
+	const queue = new PQueue({concurrency: 1});
+	queue.add(async () => {
+		await delay(400);
+		result.push('🐌');
+	}, {id: '🐌'});
+	queue.add(async () => {
+		await delay(400);
+		result.push('🦆');
+	}, {id: '🦆'});
+	queue.add(async () => {
+		await delay(400);
+		result.push('🐢');
+	}, {id: '🐢'});
+	queue.setPriority('🐢', 1);
+	await queue.onIdle();
+	t.deepEqual(result, ['🐌', '🐢', '🦆']);
+});
+
+test('track event "invoke" to check with respect to concurrency - 1', async t => {
+	const invoked: Array<string | undefined> = [];
+	const queue = new PQueue({concurrency: 1});
+	queue.on('invoke', (data: QueueAddOptions) => {
+		invoked.push(data.id);
+	});
+	const job1 = queue.add(async () => {
+		await delay(400);
+	}, {id: '🐌'});
+	const job2 = queue.add(async () => {
+		await delay(400);
+	}, {id: '🦆'});
+	t.deepEqual(invoked, ['🐌']);
+	await job1;
+	t.deepEqual(invoked, ['🐌', '🦆']);
+	await queue.onIdle();
+});