lik/ts/lik.asyncexecutionstack.ts

111 lines
3.3 KiB
TypeScript
Raw Normal View History

2023-01-18 16:45:19 +00:00
import * as plugins from './lik.plugins.js';
interface IExecutionSlot<T> {
executionDeferred: plugins.smartpromise.Deferred<T>;
funcToExecute: () => Promise<T>;
timeout?: number;
mode: 'exclusive' | 'nonexclusive';
}
2023-01-18 11:18:47 +00:00
export class AsyncExecutionStack {
2023-01-18 16:45:19 +00:00
private executionSlots: IExecutionSlot<any>[] = [];
2023-08-14 17:27:28 +00:00
private isProcessing = false;
public async getExclusiveExecutionSlot<T = any>(
funcArg: () => Promise<T>,
timeoutArg?: number
): Promise<T> {
2023-01-18 16:45:19 +00:00
const executionDeferred = plugins.smartpromise.defer<T>();
const executionSlot: IExecutionSlot<T> = {
funcToExecute: funcArg,
executionDeferred,
timeout: timeoutArg,
mode: 'exclusive',
};
this.executionSlots.push(executionSlot);
this.processExecutionSlots();
return executionDeferred.promise;
}
2023-08-14 17:27:28 +00:00
2023-01-18 16:45:19 +00:00
public async getNonExclusiveExecutionSlot<T = any>(
funcArg: () => Promise<T>,
timeoutArg?: number
2023-08-14 17:27:28 +00:00
): Promise<T> {
2023-01-18 16:45:19 +00:00
const executionDeferred = plugins.smartpromise.defer<T>();
const executionSlot: IExecutionSlot<T> = {
funcToExecute: funcArg,
executionDeferred,
timeout: timeoutArg,
mode: 'nonexclusive',
};
this.executionSlots.push(executionSlot);
this.processExecutionSlots();
return executionDeferred.promise;
}
private async processExecutionSlots() {
2023-08-14 17:27:28 +00:00
if (this.isProcessing) {
2023-01-18 16:45:19 +00:00
return;
}
2023-08-14 17:27:28 +00:00
this.isProcessing = true;
2023-01-18 16:45:19 +00:00
while (this.executionSlots.length > 0) {
2023-08-14 17:27:28 +00:00
const currentSlot = this.executionSlots[0];
if (currentSlot.mode === 'exclusive') {
await this.executeExclusiveSlot(currentSlot);
this.executionSlots.shift();
} else {
// Gather all non-exclusive slots at the front of the queue
const nonExclusiveSlots: IExecutionSlot<any>[] = [];
while (this.executionSlots.length > 0 && this.executionSlots[0].mode === 'nonexclusive') {
nonExclusiveSlots.push(this.executionSlots.shift()!);
}
await this.executeNonExclusiveSlots(nonExclusiveSlots);
}
}
this.isProcessing = false;
}
private async executeExclusiveSlot(slot: IExecutionSlot<any>) {
try {
if (slot.timeout) {
const result = await Promise.race([
slot.funcToExecute(),
plugins.smartdelay.delayFor(slot.timeout).then(() => {
throw new Error('Timeout reached');
}),
]);
slot.executionDeferred.resolve(result);
} else {
const result = await slot.funcToExecute();
slot.executionDeferred.resolve(result);
}
} catch (error) {
slot.executionDeferred.reject(error);
}
}
private async executeNonExclusiveSlots(slots: IExecutionSlot<any>[]) {
const promises = slots.map(async (slot) => {
try {
if (slot.timeout) {
2023-01-18 16:45:19 +00:00
const result = await Promise.race([
2023-08-14 17:27:28 +00:00
slot.funcToExecute(),
plugins.smartdelay.delayFor(slot.timeout).then(() => {
throw new Error('Timeout reached');
}),
2023-01-18 16:45:19 +00:00
]);
2023-08-14 17:27:28 +00:00
slot.executionDeferred.resolve(result);
2023-01-18 11:18:47 +00:00
} else {
2023-08-14 17:27:28 +00:00
const result = await slot.funcToExecute();
slot.executionDeferred.resolve(result);
2023-01-18 11:18:47 +00:00
}
2023-08-14 17:27:28 +00:00
} catch (error) {
slot.executionDeferred.reject(error);
2023-01-18 11:18:47 +00:00
}
2023-08-14 17:27:28 +00:00
});
await Promise.all(promises);
2023-01-18 16:45:19 +00:00
}
}