- Press n or j to go to the next uncovered block, b, p or k for the previous block. -
- -| 1 -2 -3 -4 -5 -6 -7 -8 -9 -10 -11 -12 -13 -14 -15 -16 -17 -18 -19 -20 -21 -22 -23 -24 -25 -26 -27 -28 -29 -30 -31 -32 -33 -34 -35 -36 -37 -38 -39 -40 -41 -42 -43 -44 -45 -46 -47 -48 -49 -50 -51 -52 -53 -54 -55 -56 -57 -58 -59 -60 -61 -62 -63 -64 -65 -66 -67 -68 -69 -70 -71 -72 -73 -74 -75 -76 -77 -78 -79 -80 -81 -82 -83 -84 -85 -86 -87 -88 -89 -90 -91 -92 -93 -94 -95 -96 -97 -98 -99 -100 -101 -102 -103 -104 -105 -106 -107 -108 -109 -110 -111 -112 -113 -114 -115 -116 -117 -118 -119 -120 -121 -122 -123 -124 -125 -126 -127 -128 -129 -130 -131 -132 -133 -134 -135 -136 -137 -138 -139 -140 -141 -142 -143 -144 -145 -146 -147 -148 -149 -150 -151 -152 -153 -154 -155 -156 -157 -158 -159 -160 -161 -162 -163 -164 -165 -166 -167 -168 -169 -170 -171 -172 -173 -174 -175 -176 -177 -178 -179 -180 -181 -182 -183 -184 -185 -186 -187 -188 -189 -190 -191 -192 -193 -194 -195 -196 -197 -198 -199 -200 -201 -202 -203 | 1x - -1x -1x - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -1x -1x - -1x - -43x - - - - - - - - - - - - - - - - - - - -43x - -43x - -43x -104x - - - -43x -43x -43x -43x -43x -43x -43x -43x -35x -43x - - -43x -5x - - -43x - - - -42x - - -27x -27x -27x -27x - - - -13x - -11x -11x - - -10x -10x - - - -3x - -3x -3x -3x - - - - - -10x - -10x -10x -10x -10x - -10x -8x -8x - -6x -6x - -1x -1x - - - -2x -2x - -1x -1x - - - -10x -8x - -2x - -10x - - - -6x -6x - - - -142x - - - -388x - - - -45x - - - -32x -32x -32x - - - - -143x - - - -27x - - - - - -1x - - - -6x - - - - - -6x - - - | import ProcessingPriorityQueue from "./ProcessingPriorityQueue";
-
-const DEFAULT_QUEUE_NAME = "default";
-const DEFAULT_CONCURRENCY_LIMIT = 1;
-
-interface PTaskOptions<T, R> {
- priority: number;
- delay?: number; // ms
- args: T;
- onRun: (args: T, execInfo?: ExecInfo) => Promise<R>;
- /**
- * Will be called when the execution of onRun is stopped based on a pause signal.
- * Used to update the arguments of the task, so that when execution resumes, the new arguments are used.
- * @param args The arguments
- * @param resSoFar The result so far returned from the execution of onRun. Null if the task hasn't yet started.
- */
- onPause?: (args: T, resSoFar: R | null) => T;
- onResume?: (args: T, resSoFar: R | null) => T;
- onCancel?: () => void;
- resultsMerge?: (resSoFar: R | null, newResult: R) => R;
- queueName?: string;
-}
-
-type ExecutionStatus =
- | "pending"
- | "running"
- | "paused"
- | "canceled"
- | "completed";
-
-export interface ExecInfo {
- getStatus: () => ExecutionStatus;
-}
-
-export class PTask<T, R> {
- private static count = 0;
-
- private static pTaskLists = new Map<string, Array<PTask<any, any>>>();
-
- private _key: number = PTask.count++;
-
- private _priority: number;
-
- private delay: number;
-
- private _onRun: (args: T, execInfo?: ExecInfo) => Promise<R>;
-
- private onPause: (args: T, resSoFar: R | null) => T;
-
- private onResume: (args: T, resSoFar: R | null) => T;
-
- private onCancel: () => void;
-
- private resultsMerge?: (resSoFar: R | null, newResult: R) => R;
-
- private queueName: string;
-
- private _args: any;
-
- private resSoFar: R | null = null;
-
- private _status: ExecutionStatus = "pending";
-
- public execInfo: ExecInfo = {
- getStatus: () => this.status,
- };
-
- constructor(options: PTaskOptions<T, R>) {
- this._priority = options.priority;
- this.delay = options.delay || 0;
- this._args = options.args;
- this._onRun = options.onRun;
- this.onPause = options.onPause || ((arg: T) => arg);
- this.onResume = options.onResume || ((arg: T) => arg);
- this.onCancel = options.onCancel || (() => {});
- this.resultsMerge =
- options.resultsMerge || ((resSoFar: R | null, newResult: R) => newResult);
- this.queueName = options.queueName || DEFAULT_QUEUE_NAME;
-
- // Add this task to the list of tasks for this queue
- if (!PTask.pTaskLists.has(this.queueName)) {
- PTask.pTaskLists.set(this.queueName, []);
- }
-
- PTask.pTaskLists.get(this.queueName).push(this);
- }
-
- public async run(): Promise<R> {
- const newRes = await ProcessingPriorityQueue.getInstance(
- this.queueName
- ).enqueue(this);
- const result = this.resultsMerge(this.resSoFar, newRes);
- this.removeSelfFromQueue();
- this._status = "completed";
- return result;
- }
-
- public async pause(): Promise<void> {
- if (this.status === "paused") return;
-
- this._status = "paused";
- const newRes = await ProcessingPriorityQueue.getInstance(
- this.queueName
- ).pause(this);
- this.resSoFar = this.resultsMerge(this.resSoFar, newRes);
- this._args = this.onPause(this.args, this.resSoFar);
- }
-
- public resume(): void {
- Iif (this.status !== "paused") return;
-
- this._status = "pending";
- this._args = this.onResume(this.args, this.resSoFar);
- ProcessingPriorityQueue.getInstance(this.queueName).resume(this);
- }
-
- public async cancel(
- { abort }: { abort: boolean } = { abort: false }
- ): Promise<[boolean, string]> {
- Iif (this.status === "canceled") return [true, "Already canceled"];
-
- const prevStatus = this._status;
- this._status = "canceled";
- let result = true;
- let message = "Successfully canceled";
-
- if (abort) {
- try {
- await ProcessingPriorityQueue.getInstance(this.queueName).cancel(this);
- } catch (err) {
- try {
- await ProcessingPriorityQueue.getInstance(this.queueName).abort(this);
- } catch {
- result = false;
- message = `${err.message}`;
- }
- }
- } else {
- try {
- await ProcessingPriorityQueue.getInstance(this.queueName).cancel(this);
- } catch (err) {
- result = false;
- message = `${err.message}`;
- }
- }
-
- if (result) {
- this.onCancel();
- } else {
- this._status = prevStatus;
- }
- return [result, message];
- }
-
- public set priority(p: number) {
- this._priority = p;
- ProcessingPriorityQueue.getInstance(this.queueName).updatePriority(this);
- }
-
- public get priority(): number {
- return this._priority;
- }
-
- public get key(): number {
- return this._key;
- }
-
- public get args(): T {
- return this._args;
- }
-
- public get onRun(): (args: T, execInfo?: ExecInfo) => Promise<R> {
- return (args: T, execInfo?: ExecInfo) => {
- this._status = "running";
- return this._onRun(args, execInfo);
- };
- }
-
- public get status(): ExecutionStatus {
- return this._status;
- }
-
- private removeSelfFromQueue(): void {
- PTask.pTaskLists
- .get(this.queueName)
- .splice(PTask.pTaskLists.get(this.queueName).indexOf(this), 1);
- }
-
- public static setConcurrencyLimit(limit: number, queueName: string = DEFAULT_QUEUE_NAME): void {
- ProcessingPriorityQueue.getInstance(queueName).setConcurrencyLimit(limit);
- }
-
- public static getAllPTasks(queueName?: string): Array<PTask<any, any>> {
- Iif (!queueName) {
- return PTask.pTaskLists.get(DEFAULT_QUEUE_NAME)
- ? [...PTask.pTaskLists.get(DEFAULT_QUEUE_NAME)]
- : []
- }
-
- return PTask.pTaskLists.get(queueName) ? [...PTask.pTaskLists.get(queueName)] : [];
- }
-}
- |
- Press n or j to go to the next uncovered block, b, p or k for the previous block. -
- -| 1 -2 -3 -4 -5 -6 -7 -8 -9 -10 -11 -12 -13 -14 -15 -16 -17 -18 -19 -20 -21 -22 -23 -24 -25 -26 -27 -28 -29 -30 -31 -32 -33 -34 -35 -36 -37 -38 -39 -40 -41 -42 -43 -44 -45 -46 -47 -48 -49 -50 -51 -52 -53 -54 -55 -56 -57 -58 -59 -60 -61 -62 -63 -64 -65 -66 -67 -68 -69 -70 -71 -72 -73 -74 -75 -76 -77 -78 -79 -80 -81 -82 -83 -84 -85 -86 -87 -88 -89 -90 -91 -92 -93 -94 -95 -96 -97 -98 -99 -100 -101 -102 -103 -104 -105 -106 -107 -108 -109 -110 -111 -112 -113 -114 -115 -116 -117 -118 -119 -120 -121 -122 -123 -124 -125 -126 -127 -128 -129 -130 -131 -132 -133 -134 -135 -136 -137 -138 -139 -140 -141 -142 -143 -144 -145 -146 -147 -148 -149 -150 -151 -152 -153 -154 -155 -156 -157 -158 -159 -160 -161 -162 -163 -164 -165 -166 -167 -168 -169 -170 -171 -172 -173 -174 -175 -176 -177 -178 -179 -180 -181 -182 -183 -184 -185 -186 -187 -188 -189 -190 -191 -192 -193 -194 -195 -196 -197 -198 -199 -200 -201 -202 -203 -204 -205 -206 -207 -208 -209 -210 -211 -212 -213 -214 -215 -216 -217 -218 -219 -220 -221 -222 -223 -224 -225 -226 | 2x - -2x - - - - - - - - - -2x - - - -2x - - - - -7x - - - - -7x - - - - -7x - - - - - - -7x - -7x -7x - - -71x - - - -1x - - - -42x -40x -40x -40x -40x - - -42x -42x -42x - - -42x - - - -49x -49x - -49x - - - -6x - -6x -6x - -6x -6x -6x - - - -10x - -3x -3x -3x -3x - - - - -6x -4x - - -2x -1x - -1x - - - -1x -1x - -1x - - - -4x -4x -4x - - -4x - - - - - - - -11x - -2x -2x - -2x -2x -2x - - - - - -9x -8x -8x -8x -8x - -8x - - -1x - - - -3x - -3x -3x - - -3x -3x -3x - - -3x - - - -96x - -49x - -49x -49x - - -96x -49x - -49x -49x - -49x -17x -17x - - - -32x - - -31x - - - - - - - - -1x - - - - - - - - -32x - - - - - -83x -7x - - - - - -83x - - - | import FastPriorityQueue from "fastpriorityqueue";
-import { PTask } from "./PTask";
-import SubscribableQueueItem from "./SubscribableQueueItem";
-
-interface PriorityQueueItem<T> {
- value: T;
- valid: boolean;
-}
-
-/**
- * Singleton class that manages a queue of tasks.
- */
-export default class ProcessingPriorityQueue {
- /**
- * A map of all the instances of the queue.
- */
- private static instances = new Map<string, ProcessingPriorityQueue>();
-
- /**
- * This map is used to keep track of the paused tasks.
- */
- private pausedTasks = new Map<number, SubscribableQueueItem>();
-
- /**
- * Holds the currently running tasks.
- */
- private currentlyRunning = new Map<number, SubscribableQueueItem>();
-
- /**
- * This map is used to keep track of the tasks waiting for execution.
- */
- private pendingTasks = new Map<number, PriorityQueueItem<SubscribableQueueItem>>();
-
- /**
- * Holds a prioritized heap of the pending tasks.
- */
- private priorityQueue: FastPriorityQueue<PriorityQueueItem<SubscribableQueueItem>>;
-
- private currentConcurrencyCount = 0;
-
- private constructor(private concurrencyLimit: number = 1) {
- this.priorityQueue = new FastPriorityQueue((
- a: PriorityQueueItem<SubscribableQueueItem>,
- b: PriorityQueueItem<SubscribableQueueItem>
- ) => a.value.task.priority > b.value.task.priority);
- }
-
- public setConcurrencyLimit(concurrencyLimit: number): void {
- this.concurrencyLimit = concurrencyLimit;
- }
-
- public async enqueue(ptask: PTask<any, any>): Promise<any> {
- if (!this.pendingTasks.has(ptask.key)) {
- const subscribableItem = new SubscribableQueueItem(ptask);
- const qItem = { value: subscribableItem, valid: true };
- this.pendingTasks.set(ptask.key, qItem);
- this.priorityQueue.add(qItem);
- }
-
- const promise = this.pendingTasks.get(ptask.key).value.createPromise();
- setImmediate(() => {
- this.process();
- });
-
- return promise;
- }
-
- private poll(): PriorityQueueItem<SubscribableQueueItem> {
- const item = this.priorityQueue.poll() as PriorityQueueItem<SubscribableQueueItem>;
- this.pendingTasks.delete(item.value.task.key);
-
- return item;
- }
-
- public updatePriority(ptask: PTask<any, any>): void {
- Iif(!this.pendingTasks.has(ptask.key)) return;
-
- const qItem = this.pendingTasks.get(ptask.key);
- qItem.valid = false;
-
- const newQItem = { value: qItem.value, valid: true }
- this.pendingTasks.set(ptask.key, newQItem);
- this.priorityQueue.add(newQItem);
- }
-
- public async cancel(ptask: PTask<any, any>): Promise<boolean> {
- if (!this.pendingTasks.has(ptask.key)) throw new Error("Task not found");
-
- const qItem = this.pendingTasks.get(ptask.key);
- qItem.valid = false;
- qItem.value.rejectCallback(new Error("Task canceled"), "eventual");
- return true;
- }
-
- public async abort(ptask: PTask<any, any>): Promise<any> {
- // get the item and remove it from currently running
- if (this.currentlyRunning.has(ptask.key))
- return this.abortRunning(this.currentlyRunning.get(ptask.key));
-
- // check if task is paused
- if (this.pausedTasks.has(ptask.key))
- return this.abortPaused(this.pausedTasks.get(ptask.key));
-
- throw new Error("Cannot abort a task that is not running");
- }
-
- private async abortPaused(queueItem: SubscribableQueueItem): Promise<any> {
- this.pausedTasks.delete(queueItem.task.key);
- queueItem.rejectCallback(new Error("Paused task aborted"), "eventual");
-
- return true;
- }
-
- private async abortRunning(queueItem: SubscribableQueueItem): Promise<any> {
- this.currentlyRunning.delete(queueItem.task.key);
- try {
- await queueItem.createPromise("immediate");
- } catch {
- } finally {
- queueItem.rejectCallback(new Error("Running task aborted"), "eventual");
- }
- }
-
- public async pause(ptask: PTask<any, any>): Promise<any> {
- /**
- * Check if the task is currently running
- */
- if (this.currentlyRunning.has(ptask.key)) {
- // TODO: THIS PROBABLY DON'T WORK - Actually might work but cause a memory leak
- const currentlyRunningTask = this.currentlyRunning.get(ptask.key);
- this.pausedTasks.set(currentlyRunningTask.task.key, currentlyRunningTask);
-
- const result = await currentlyRunningTask.createPromise("immediate");
- this.process();
- return result;
- }
-
- /**
- * Check if the task hasn't started yet
- */
- if (this.pendingTasks.has(ptask.key)) {
- const qItem = this.pendingTasks.get(ptask.key);
- qItem.valid = false;
- this.pausedTasks.set(ptask.key, this.pendingTasks.get(ptask.key).value);
- this.pendingTasks.delete(ptask.key);
-
- return null;
- }
-
- throw new Error("Cannot pause a task that is not running");
- }
-
- public async resume(ptask: PTask<any, any>): Promise<any> {
- Iif (!this.pausedTasks.has(ptask.key)) return null;
-
- const pausedItem = this.pausedTasks.get(ptask.key);
- this.pausedTasks.delete(pausedItem.task.key);
-
- // Add the task to the priority queue
- const qItem = { value: pausedItem, valid: true };
- this.pendingTasks.set(pausedItem.task.key, qItem);
- this.priorityQueue.add(qItem);
-
- // resume the task
- setImmediate(() => this.process());
- }
-
- private async process(): Promise<void> {
- const proceedWithNextItem = (curTask: SubscribableQueueItem) => {
- // remove from currently running
- this.currentlyRunning.delete(curTask.task.key);
-
- this.currentConcurrencyCount--;
- this.process();
- }
-
- if (this.currentConcurrencyCount < this.concurrencyLimit && this.priorityQueue.size > 0) {
- this.currentConcurrencyCount++;
-
- const {value: sqItem, valid} = this.poll();
- this.currentlyRunning.set(sqItem.task.key, sqItem);
-
- if (!valid) {
- proceedWithNextItem(sqItem);
- return;
- }
-
- /* Process the item with the given procedure */
- sqItem.task
- .onRun(sqItem.task.args, sqItem.task.execInfo)
- .then((result) => {
- sqItem.resolveCallback(
- result,
- sqItem.task.execInfo.getStatus() === "canceled" ||
- sqItem.task.execInfo.getStatus() === "paused"
- ? "immediate"
- : "eventual"
- );
- })
- .catch((error) => {
- sqItem.rejectCallback(
- error,
- sqItem.task.execInfo.getStatus() === "canceled" ||
- sqItem.task.execInfo.getStatus() === "paused"
- ? "immediate"
- : "eventual"
- );
- })
- .finally(async () => {
- proceedWithNextItem(sqItem);
- });
- }
- }
-
- static getInstance(queueName: string): ProcessingPriorityQueue {
- if (!ProcessingPriorityQueue.instances.has(queueName)) {
- ProcessingPriorityQueue.instances.set(
- queueName,
- new ProcessingPriorityQueue()
- );
- }
-
- return ProcessingPriorityQueue.instances.get(queueName);
- }
-}
- |
- Press n or j to go to the next uncovered block, b, p or k for the previous block. -
- -| 1 -2 -3 -4 -5 -6 -7 -8 -9 -10 -11 -12 -13 -14 -15 -16 -17 -18 -19 -20 -21 -22 -23 -24 -25 -26 -27 -28 -29 -30 -31 -32 -33 -34 -35 -36 -37 -38 -39 -40 -41 -42 -43 -44 -45 -46 -47 -48 -49 -50 -51 -52 -53 -54 -55 -56 -57 -58 -59 -60 -61 -62 -63 -64 -65 -66 -67 -68 -69 -70 -71 -72 -73 -74 -75 -76 | - - - - - - - -2x -40x - - - - -40x - - - - -40x - - -31x -34x - - - - -9x -9x - - - - - - - -48x -48x -34x - - -48x -9x - - - - -48x -48x - -48x - - - - - - - - - - - - -96x - -48x -48x - - -48x -48x - - - - | import { PTask } from "./PTask";
-
-type ListenerType = "eventual" | "immediate";
-
-/**
- * A queue item that can be processed by a processing queue.
- * It keeps a list of callbacks that can be called when the item is processed.
- */
-export default class SubscribableQueueItem {
- private rejectCallbacks = {
- eventual: new Array<Function>(),
- immediate: new Array<Function>(),
- };
-
- private resolveCallbacks = {
- eventual: new Array<Function>(),
- immediate: new Array<Function>(),
- };
-
- constructor(public task: PTask<any, any>) {}
-
- public resolveCallback(result: any, type: ListenerType) {
- this.resolveCallbacks[type].forEach((resolveCB: Function) => {
- resolveCB(result);
- });
- }
-
- public rejectCallback(error: any, type: ListenerType) {
- this.rejectCallbacks[type].forEach((rejectCB: Function) => {
- rejectCB(error);
- });
- }
-
- public createPromise(type: ListenerType = "eventual"): Promise<any> {
- // Prepare the promise
- let resolveCallback: Function;
- let rejectCallback: Function;
- const promise = new Promise((resolve, reject) => {
- resolveCallback = async (result) => {
- resolve(result);
- };
-
- rejectCallback = (error: Error) => {
- reject(error);
- };
- });
-
- // Prepare queue item
- this.subscribe("resolve", type, resolveCallback);
- this.subscribe("reject", type, rejectCallback);
-
- return promise;
- }
-
- /**
- * Set the callback functions called when the request item gets resolved or rejected
- * @param eventType The type of action to set callback for
- * @param callback The callback fot the action type
- */
- private subscribe(
- eventType: "resolve" | "reject",
- listenerType: ListenerType,
- callback: Function
- ) {
- switch (eventType) {
- case "resolve":
- this.resolveCallbacks[listenerType].push(callback);
- break;
-
- case "reject":
- this.rejectCallbacks[listenerType].push(callback);
- break;
- }
- }
-}
- |
- Press n or j to go to the next uncovered block, b, p or k for the previous block. -
- -| File | -- | Statements | -- | Branches | -- | Functions | -- | Lines | -- |
|---|---|---|---|---|---|---|---|---|---|
| PTask.ts | -
-
- |
- 96.1% | -74/77 | -74.07% | -20/27 | -100% | -20/20 | -98.61% | -71/72 | -
| ProcessingPriorityQueue.ts | -
-
- |
- 97.75% | -87/89 | -86.36% | -19/22 | -100% | -20/20 | -100% | -85/85 | -
| SubscribableQueueItem.ts | -
-
- |
- 100% | -21/21 | -100% | -3/3 | -100% | -10/10 | -100% | -21/21 | -
| index.ts | -
-
- |
- 100% | -2/2 | -100% | -0/0 | -100% | -1/1 | -100% | -2/2 | -
- Press n or j to go to the next uncovered block, b, p or k for the previous block. -
- -| 1 -2 -3 | 1x -52x - | import { PTask, ExecInfo } from "./PTask";
-export { PTask, ExecInfo };
- |