Feature/update priority

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' | 'started';
 
 /**
 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 */
+	#uidAssigner = 1;
+
 	/**
 	Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
 
@@ -228,15 +231,25 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 		});
 	}
 
+	async setPriority(uid: string, priority: number) {
+		this.#queue.prioritize(uid, 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> {
+	async add<TaskResultType>(function_: Task<TaskResultType>, options: {throwOnTimeout: true} & Exclude<EnqueueOptionsType, 'throwOnTimeout'>, uid?: string): Promise<TaskResultType>;
+	async add<TaskResultType>(function_: Task<TaskResultType>, options?: Partial<EnqueueOptionsType>, uid?: string): Promise<TaskResultType | void>;
+	async add<TaskResultType>(function_: Task<TaskResultType>, options: Partial<EnqueueOptionsType> = {}, uid?: string): Promise<TaskResultType | void> {
+		// Incase uid is not defined
+		if (uid === undefined) {
+			uid = (this.#uidAssigner++).toString();
+		}
+
 		options = {
 			timeout: this.timeout,
 			throwOnTimeout: this.#throwOnTimeout,
+			uid,
 			...options,
 		};
 
@@ -258,6 +271,7 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 						operation = Promise.race([operation, this.#throwOnAbort(options.signal)]);
 					}
 
+					this.emit('started', uid);
 					const result = await operation;
 					resolve(result);
 					this.emit('completed', result);

source/priority-queue.ts

@@ -4,6 +4,7 @@ import {type QueueAddOptions} from './options.js';
 
 export type PriorityQueueOptions = {
 	priority?: number;
+	uid?: string;
 } & QueueAddOptions;
 
 export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOptions> {
@@ -32,6 +33,27 @@ export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOp
 		this.#queue.splice(index, 0, element);
 	}
 
+	prioritize(uid: string, priority?: number) {
+		const queueIndex: number = this.#queue.findIndex((element: Readonly<PriorityQueueOptions>) => element.uid === uid);
+		const [item] = this.#queue.splice(queueIndex, 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;
+	prioritize: (uid: string, priority: number) => void;
 };

test/test.ts

@@ -6,7 +6,7 @@ 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';
 
 const fixture = Symbol('fixture');
 
@@ -1134,3 +1134,47 @@ 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('🐌');
+	}, {}, 'snail');
+	queue.add(async () => {
+		await delay(400);
+		result.push('🦆');
+	}, {}, 'duck');
+	queue.add(async () => {
+		await delay(400);
+		result.push('🐢');
+	}, {}, 'turtle');
+	queue.setPriority('turtle', 1);
+	await queue.onIdle();
+	t.deepEqual(result, ['🐌', '🐢', '🦆']);
+});
+
+test('started event to check when promise function is called', async t => {
+	const result: string[] = [];
+	const queue = new PQueue({concurrency: 1});
+	queue.add(async () => {
+		await delay(400);
+		result.push('🐌');
+	}, {}, '🐌');
+	queue.add(async () => {
+		await delay(400);
+		result.push('🦆');
+	}, {}, '🦆');
+	queue.add(async () => {
+		await delay(400);
+		result.push('🐢');
+	}, {}, '🐢');
+	queue.on('started', uid => {
+		if (uid === '🦆') {
+			t.deepEqual(result, ['🐌', '🐢']);
+		}
+	});
+	queue.setPriority('🐢', 1);
+	await queue.onIdle();
+});