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 */
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
/** Use to assign a unique identifier to a promise function, if not explicitly specified */
// Used to assign a unique identifier to a promise function, if not explicitly specified.

#idAssigner = 1;
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should be a BigInt, to ensure it will never overflow.


/**
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) {
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Doc comment

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
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
// Incase id is not defined
// In case `id` is not defined.

if (options.id === undefined) {
options.id = (this.#idAssigner++).toString();
}
Comment on lines +245 to +247
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if (options.id === undefined) {
options.id = (this.#idAssigner++).toString();
}
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;
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Doc comment.

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
id?: string;
readonly id?: string;

} & TaskOptions & TimeoutOptions;

export type TaskOptions = {
Expand Down
17 changes: 16 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,20 @@ 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);
if (item === undefined) {
throw new Error('Undefined Item - No promise function of specified id available in the queue.');
}
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think these should throw. It should be allowed to set a priority without knowing beforehand whether the has already finished.


this.enqueue(item.run, {priority, id});
}

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 => {
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

test.failing is not for tests that are intentionally failing.

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, ['🐌', '🐢', '🦆']);
});