fix(core): update

This commit is contained in:
Philipp Kunz 2023-08-12 12:24:10 +02:00
parent fcd7ea467e
commit bae776d4e9
3 changed files with 79 additions and 123 deletions

View File

@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@push.rocks/taskbuffer', name: '@push.rocks/taskbuffer',
version: '3.1.1', version: '3.1.2',
description: 'flexible task management. TypeScript ready!' description: 'flexible task management. TypeScript ready!'
} }

View File

@ -2,13 +2,10 @@ import { Task } from './taskbuffer.classes.task.js';
import * as plugins from './taskbuffer.plugins.js'; import * as plugins from './taskbuffer.plugins.js';
/** /**
* constains all data for the final coordinator to actually make an informed decision * Contains all data for the final coordinator to make an informed decision.
*/ */
export interface IDistributedTaskRequest { export interface IDistributedTaskRequest {
/** submitterId: string;
* this needs to correlate to the consultationResult
*/
submitterRandomId: string;
taskName: string; taskName: string;
taskVersion: string; taskVersion: string;
taskExecutionTime: number; taskExecutionTime: number;
@ -18,24 +15,19 @@ export interface IDistributedTaskRequest {
} }
export interface IDistributedTaskRequestResult { export interface IDistributedTaskRequestResult {
/** submitterId: string;
* this needs to correlate to the decisionInfoBasis
*/
submitterRandomId: string;
/**
* can be used while debugging
*/
considered: boolean; considered: boolean;
rank: string; rank: number;
reason: string; reason: string;
shouldTrigger: boolean; shouldTrigger: boolean;
} }
export abstract class AbstractDistributedCoordinator { export abstract class AbstractDistributedCoordinator {
public abstract fireDistributedTaskRequest( public abstract fireDistributedTaskRequest(
infoBasisArg: IDistributedTaskRequest infoBasis: IDistributedTaskRequest
): Promise<IDistributedTaskRequestResult>; ): Promise<IDistributedTaskRequestResult>;
public abstract updateDistributedTaskRequest( public abstract updateDistributedTaskRequest(
infoBasisArg: IDistributedTaskRequest infoBasis: IDistributedTaskRequest
): Promise<void>; ): Promise<void>;
} }

View File

@ -1,10 +1,10 @@
import * as plugins from './taskbuffer.plugins.js'; import * as plugins from './taskbuffer.plugins.js';
import { Task } from './taskbuffer.classes.task.js'; import { Task } from './taskbuffer.classes.task.js';
import { AbstractDistributedCoordinator } from './taskbuffer.classes.distributedcoordinator.js'; import { AbstractDistributedCoordinator, type IDistributedTaskRequestResult } from './taskbuffer.classes.distributedcoordinator.js';
export interface ICronJob { export interface ICronJob {
cronString: string; cronString: string;
taskNameArg: string; taskName: string;
job: any; job: any;
} }
@ -16,54 +16,34 @@ export class TaskManager {
public randomId = plugins.isounique.uni(); public randomId = plugins.isounique.uni();
public taskMap = new plugins.lik.ObjectMap<Task>(); public taskMap = new plugins.lik.ObjectMap<Task>();
private cronJobManager = new plugins.smarttime.CronManager(); private cronJobManager = new plugins.smarttime.CronManager();
public options: ITaskManagerConstructorOptions = { public options: ITaskManagerConstructorOptions = {
distributedCoordinator: null, distributedCoordinator: null,
}; };
constructor(optionsArg: ITaskManagerConstructorOptions = {}) { constructor(options: ITaskManagerConstructorOptions = {}) {
this.options = Object.assign(this.options, optionsArg); this.options = Object.assign(this.options, options);
} }
/** public getTaskByName(taskName: string): Task {
* checks if a task is already present return this.taskMap.findSync((task) => task.name === taskName);
* @param taskNameArg
*/
public getTaskByName(taskNameArg: string): Task {
return this.taskMap.findSync((itemArg) => {
return itemArg.name === taskNameArg;
});
} }
/** public addTask(task: Task): void {
* adds a Task to the TaskManager if (!task.name) {
* @param taskArg throw new Error('Task must have a name to be added to taskManager');
*/
public addTask(taskArg: Task): void {
if (!taskArg.name) {
throw new Error('taskArg needs a name to be added to taskManager');
} }
this.taskMap.add(taskArg); this.taskMap.add(task);
} }
/** public addAndScheduleTask(task: Task, cronString: string) {
* adds and schedules a task at once this.addTask(task);
* @param taskArg this.scheduleTaskByName(task.name, cronString);
* @param cronStringArg
*/
public addAndScheduleTask(taskArg: Task, cronStringArg: string) {
this.addTask(taskArg);
this.scheduleTaskByName(taskArg.name, cronStringArg);
} }
/** public async triggerTaskByName(taskName: string): Promise<any> {
* triggers a task in the TaskManagerByName const taskToTrigger = this.getTaskByName(taskName);
* @param taskNameArg
*/
public triggerTaskByName(taskNameArg: string): Promise<any> {
const taskToTrigger = this.getTaskByName(taskNameArg);
if (!taskToTrigger) { if (!taskToTrigger) {
throw new Error(`There is no task with the name of ${taskNameArg}`); throw new Error(`No task with the name ${taskName} found.`);
} }
return taskToTrigger.trigger(); return taskToTrigger.trigger();
} }
@ -72,95 +52,79 @@ export class TaskManager {
return task.trigger(); return task.trigger();
} }
/** public scheduleTaskByName(taskName: string, cronString: string) {
* schedules the task by name const taskToSchedule = this.getTaskByName(taskName);
* @param taskNameArg if (!taskToSchedule) {
*/ throw new Error(`No task with the name ${taskName} found.`);
public scheduleTaskByName(taskNameArg: string, cronStringArg: string) { }
const taskToSchedule = this.getTaskByName(taskNameArg); this.handleTaskScheduling(taskToSchedule, cronString);
const cronJob = this.cronJobManager.addCronjob( }
cronStringArg,
async (triggerTimeArg: number) => {
console.log(`taskbuffer schedule triggered task >>${taskToSchedule.name}<<`);
console.log(
`task >>${taskToSchedule.name}<< is ${
taskToSchedule.buffered
? `buffered with max ${taskToSchedule.bufferMax} buffered calls`
: `unbuffered`
}`
);
if (this.options.distributedCoordinator) {
console.log(`Found a distrubuted coordinator, performing distributed consultation.`);
const announcementResult =
await this.options.distributedCoordinator.fireDistributedTaskRequest({
submitterRandomId: this.randomId,
status: 'requesting',
taskExecutionParallel: 1,
taskExecutionTime: triggerTimeArg,
taskExecutionTimeout: taskToSchedule.timeout,
taskName: taskToSchedule.name,
taskVersion: taskToSchedule.version,
});
private handleTaskScheduling(task: Task, cronString: string) {
const cronJob = this.cronJobManager.addCronjob(
cronString,
async (triggerTime: number) => {
this.logTaskState(task);
if (this.options.distributedCoordinator) {
const announcementResult = await this.performDistributedConsultation(task, triggerTime);
if (!announcementResult.shouldTrigger) { if (!announcementResult.shouldTrigger) {
console.log('distributed coordinator result: NOT EXECUTING'); console.log('Distributed coordinator result: NOT EXECUTING');
return; return;
} else { } else {
console.log('distributed coordinator result: CHOSEN AND EXECUTING'); console.log('Distributed coordinator result: CHOSEN AND EXECUTING');
} }
} }
await taskToSchedule.trigger(); await task.trigger();
} }
); );
taskToSchedule.cronJob = cronJob; task.cronJob = cronJob;
} }
/** private logTaskState(task: Task) {
* deschedules a task by name console.log(`Taskbuffer schedule triggered task >>${task.name}<<`);
* @param taskNameArg const bufferState = task.buffered
*/ ? `buffered with max ${task.bufferMax} buffered calls`
public descheduleTaskByName(taskNameArg: string) { : `unbuffered`;
const taskToDeSchedule = this.getTaskByName(taskNameArg); console.log(`Task >>${task.name}<< is ${bufferState}`);
if (taskToDeSchedule.cronJob) { }
this.cronJobManager.removeCronjob(taskToDeSchedule.cronJob);
taskToDeSchedule.cronJob = null; private async performDistributedConsultation(task: Task, triggerTime: number): Promise<IDistributedTaskRequestResult> {
console.log('Found a distributed coordinator, performing consultation.');
return this.options.distributedCoordinator.fireDistributedTaskRequest({
submitterId: this.randomId,
status: 'requesting',
taskExecutionParallel: 1,
taskExecutionTime: triggerTime,
taskExecutionTimeout: task.timeout,
taskName: task.name,
taskVersion: task.version,
});
}
public descheduleTaskByName(taskName: string) {
const task = this.getTaskByName(taskName);
if (task && task.cronJob) {
this.cronJobManager.removeCronjob(task.cronJob);
task.cronJob = null;
} }
if (this.cronJobManager.cronjobs.isEmpty) { if (this.cronJobManager.cronjobs.isEmpty) {
this.cronJobManager.stop(); this.cronJobManager.stop();
} }
} }
/**
* deschedules a task
* @param task
*/
public async descheduleTask(task: Task) { public async descheduleTask(task: Task) {
await this.descheduleTaskByName(task.name); await this.descheduleTaskByName(task.name);
} }
/** public getScheduleForTaskName(taskName: string): string | null {
* returns the schedule of a specific task const task = this.getTaskByName(taskName);
* @param taskNameArg return task && task.cronJob ? task.cronJob.cronExpression : null;
*/
public getScheduleForTaskName(taskNameArg: string): string | null {
const task = this.getTaskByName(taskNameArg);
if (!task || !task.cronJob) {
return null;
} }
return task.cronJob.cronExpression;
}
/**
* starts the taskmanager
*/
public start() { public start() {
this.cronJobManager.start(); this.cronJobManager.start();
} }
/**
* stops the taskmanager
*/
public stop() { public stop() {
this.cronJobManager.stop(); this.cronJobManager.stop();
} }