3.1.3

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/taskbuffer",
-  "version": "3.0.14",
+  "version": "3.1.3",
   "private": false,
   "description": "flexible task management. TypeScript ready!",
   "main": "dist_ts/index.js",
@@ -42,8 +42,8 @@
     "@gitzone/tsbundle": "^2.0.8",
     "@gitzone/tsrun": "^1.2.44",
     "@gitzone/tstest": "^1.0.77",
-    "@push.rocks/tapbundle": "^5.0.12",
+    "@push.rocks/tapbundle": "^5.0.15",
-    "@types/node": "^20.4.5"
+    "@types/node": "^20.4.10"
   },
   "files": [
     "ts/**/*",

test/test.1.task.ts

@@ -12,13 +12,6 @@ tap.test('new Task() should return a new task', async () => {
   expect(testTask).toBeInstanceOf(taskbuffer.Task);
 });
 
-tap.test('should be able to get the task state', async () => {
-  const testTask = new taskbuffer.Task({
-    taskFunction: async () => {},
-  });
-  expect(testTask.state).toEqual('ready');
-});
-
 tap.test('should have bufferMax set to the provided value', async () => {
   const task2 = new taskbuffer.Task({
     taskFunction: async () => {},

ts/00_commitinfo_data.ts

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

ts/taskbuffer.classes.distributedcoordinator.ts

@@ -2,13 +2,10 @@ import { Task } from './taskbuffer.classes.task.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 {
-  /**
+  submitterId: string;
-   * this needs to correlate to the consultationResult
-   */
-  submitterRandomId: string;
   taskName: string;
   taskVersion: string;
   taskExecutionTime: number;
@@ -18,24 +15,22 @@ export interface IDistributedTaskRequest {
 }
 
 export interface IDistributedTaskRequestResult {
-  /**
+  submitterId: string;
-   * this needs to correlate to the decisionInfoBasis
-   */
-  submitterRandomId: string;
-  /**
-   * can be used while debugging
-   */
   considered: boolean;
-  rank: string;
+  rank: number;
   reason: string;
   shouldTrigger: boolean;
 }
 
 export abstract class AbstractDistributedCoordinator {
   public abstract fireDistributedTaskRequest(
-    infoBasisArg: IDistributedTaskRequest
+    infoBasis: IDistributedTaskRequest
   ): Promise<IDistributedTaskRequestResult>;
+
   public abstract updateDistributedTaskRequest(
-    infoBasisArg: IDistributedTaskRequest
+    infoBasis: IDistributedTaskRequest
   ): Promise<void>;
+
+  public abstract start(): Promise<void>;
+  public abstract stop(): Promise<void>;
 }

ts/taskbuffer.classes.task.ts

@@ -15,8 +15,9 @@ export interface ITaskSetupFunction<T = undefined> {
 export type TPreOrAfterTaskFunction = () => Task<any>;
 
 export class Task<T = undefined> {
-  // STATIC
+  public static extractTask<T = undefined>(
-  public static extractTask<T = undefined>(preOrAfterTaskArg: Task<T> | TPreOrAfterTaskFunction): Task<T> {
+    preOrAfterTaskArg: Task<T> | TPreOrAfterTaskFunction
+  ): Task<T> {
     switch (true) {
       case !preOrAfterTaskArg:
         return null;
@@ -44,7 +45,7 @@ export class Task<T = undefined> {
     }
   };
 
-  public static isTaskTouched<T = undefined> (
+  public static isTaskTouched<T = undefined>(
     taskArg: Task<T> | TPreOrAfterTaskFunction,
     touchedTasksArray: Task<T>[]
   ): boolean {
@@ -56,7 +57,7 @@ export class Task<T = undefined> {
       }
     }
     return result;
-  };
+  }
 
   public static runTask = async <T>(
     taskArg: Task<T> | TPreOrAfterTaskFunction,
@@ -65,6 +66,11 @@ export class Task<T = undefined> {
     const taskToRun = Task.extractTask(taskArg);
     const done = plugins.smartpromise.defer();
 
+    // Wait for all blocking tasks to finish
+    for (const task of taskToRun.blockingTasks) {
+      await task.finished;
+    }
+
     if (!taskToRun.setupValue && taskToRun.taskSetup) {
       taskToRun.setupValue = await taskToRun.taskSetup();
     }
@@ -77,6 +83,14 @@ export class Task<T = undefined> {
 
     done.promise.then(async () => {
       taskToRun.running = false;
+
+      // When the task has finished running, resolve the finished promise
+      taskToRun.resolveFinished();
+
+      // Create a new finished promise for the next run
+      taskToRun.finished = new Promise((resolve) => {
+        taskToRun.resolveFinished = resolve;
+      });
     });
 
     const options = {
@@ -125,7 +139,6 @@ export class Task<T = undefined> {
     return await done.promise;
   };
 
-  // INSTANCE
   public name: string;
   public version: string;
   public taskFunction: ITaskFunction<T>;
@@ -139,12 +152,20 @@ export class Task<T = undefined> {
   public preTask: Task<T> | TPreOrAfterTaskFunction;
   public afterTask: Task<T> | TPreOrAfterTaskFunction;
 
+  // Add a list to store the blocking tasks
+  public blockingTasks: Task[] = [];
+
+  // Add a promise that will resolve when the task has finished
+  private finished: Promise<void>;
+  private resolveFinished: () => void;
+
   public running: boolean = false;
   public bufferRunner = new BufferRunner(this);
   public cycleCounter = new CycleCounter(this);
 
-  public idle: boolean = true;
+  public get idle() {
-  private _state: string = 'ready';
+    return !this.running;
+  }
 
   public taskSetup: ITaskSetupFunction<T>;
   public setupValue: T;
@@ -162,12 +183,16 @@ export class Task<T = undefined> {
     this.taskFunction = optionsArg.taskFunction;
     this.preTask = optionsArg.preTask;
     this.afterTask = optionsArg.afterTask;
-    this.idle = !this.running;
     this.buffered = optionsArg.buffered;
     this.bufferMax = optionsArg.bufferMax;
     this.execDelay = optionsArg.execDelay;
     this.name = optionsArg.name;
     this.taskSetup = optionsArg.taskSetup;
+
+    // Create the finished promise
+    this.finished = new Promise((resolve) => {
+      this.resolveFinished = resolve;
+    });
   }
 
   public trigger(x?: any): Promise<any> {
@@ -185,16 +210,4 @@ export class Task<T = undefined> {
   public triggerBuffered(x?: any): Promise<any> {
     return this.bufferRunner.trigger(x);
   }
-
-  get state(): string {
-    return this._state;
-  }
-
-  set state(stateArg: string) {
-    if (stateArg === 'locked') {
-      this._state = 'locked';
-    } else {
-      logger.log('error', `state type ${stateArg} could not be set`);
-    }
-  }
 }

ts/taskbuffer.classes.taskmanager.ts

@@ -1,10 +1,10 @@
 import * as plugins from './taskbuffer.plugins.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 {
   cronString: string;
-  taskNameArg: string;
+  taskName: string;
   job: any;
 }
 
@@ -16,54 +16,34 @@ export class TaskManager {
   public randomId = plugins.isounique.uni();
   public taskMap = new plugins.lik.ObjectMap<Task>();
   private cronJobManager = new plugins.smarttime.CronManager();
-
   public options: ITaskManagerConstructorOptions = {
     distributedCoordinator: null,
   };
 
-  constructor(optionosArg: ITaskManagerConstructorOptions = {}) {
+  constructor(options: ITaskManagerConstructorOptions = {}) {
-    this.options = Object.assign(this.options, optionosArg);
+    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) {
-      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();
   }
@@ -72,88 +52,86 @@ export class TaskManager {
     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) {
-            console.log('distributed coordinator result: NOT EXECUTING');
+            console.log('Distributed coordinator result: NOT EXECUTING');
             return;
           } 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) {
       this.cronJobManager.stop();
     }
   }
 
-  /**
-   * deschedules a task
-   * @param task
-   */
   public async descheduleTask(task: Task) {
     await this.descheduleTaskByName(task.name);
   }
-  /**
-   * returns all schedules of a specific task
-   * @param taskNameArg
-   */
-  public getSchedulesForTaskName(taskNameArg: string) {}
 
-  /**
+  public getScheduleForTaskName(taskName: string): string | null {
-   * starts the taskmanager
+    const task = this.getTaskByName(taskName);
-   */
+    return task && task.cronJob ? task.cronJob.cronExpression : null;
-  public start() {
+  }
+
+  public async start() {
+    if (this.options.distributedCoordinator) {
+      await this.options.distributedCoordinator.start();
+    }
     this.cronJobManager.start();
   }
 
-  /**
+  public async stop() {
-   * stops the taskmanager
-   */
-  public stop() {
     this.cronJobManager.stop();
+    if (this.options.distributedCoordinator) {
+      await this.options.distributedCoordinator.stop();
+    }
   }
 }