Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature/update priority #209

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
12 changes: 12 additions & 0 deletions source/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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,
Expand Down
1 change: 1 addition & 0 deletions source/options.ts
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ export type QueueAddOptions = {
@default 0
*/
readonly priority?: number;
id?: string;
} & TaskOptions & TimeoutOptions;

export type TaskOptions = {
Expand Down
28 changes: 27 additions & 1 deletion source/priority-queue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,11 @@ export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOp

const element = {
priority: options.priority,
id: options.id,
run,
};

if (this.size && this.#queue[this.size - 1]!.priority! >= options.priority!) {
if (this.size === 0 || this.#queue[this.size - 1]!.priority! >= options.priority!) {
this.#queue.push(element);
return;
}
Expand All @@ -32,6 +33,31 @@ 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);
if (existingIndex === -1) {
throw new Error('Invalid Index - No promise function of specified id available in the queue.');
}

const [item] = this.#queue.splice(existingIndex, 1);
RaishavHanspal marked this conversation as resolved.
Show resolved Hide resolved
if (item === undefined) {
return;
RaishavHanspal marked this conversation as resolved.
Show resolved Hide resolved
}
RaishavHanspal marked this conversation as resolved.
Show resolved Hide resolved

item.priority = priority;
if (this.size === 0 || 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;
Expand Down
1 change: 1 addition & 0 deletions source/queue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
};
42 changes: 41 additions & 1 deletion test/test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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');

Expand Down Expand Up @@ -1134,3 +1134,43 @@ 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.failing('.setPriority() - with invalid "id"', 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, ['🐌', '🐢', '🦆']);
});