X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=de65f27012634c493d697a0ae9e3972a7f5234db;hb=5aa31a743fde300612ae89a242b809ae7db083ed;hp=6eb2d646bdcaa86f93eb07cf341012463f295a9a;hpb=3aa16da67d1de7ed79de0fa4eedccc9cc7f58a52;p=poolifier.git

diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts
index 6eb2d646..de65f270 100644
--- a/src/pools/worker-node.ts
+++ b/src/pools/worker-node.ts
@@ -15,7 +15,7 @@ import {
   type IWorkerNode,
   type StrategyData,
   type WorkerInfo,
-  type WorkerNodeEventCallback,
+  type WorkerNodeEventDetail,
   type WorkerType,
   WorkerTypes,
   type WorkerUsage
@@ -29,7 +29,8 @@ import { checkWorkerNodeArguments } from './utils'
  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  */
 export class WorkerNode<Worker extends IWorker, Data = unknown>
-implements IWorkerNode<Worker, Data> {
+  extends EventTarget
+  implements IWorkerNode<Worker, Data> {
   /** @inheritdoc */
   public readonly worker: Worker
   /** @inheritdoc */
@@ -42,10 +43,6 @@ implements IWorkerNode<Worker, Data> {
   public messageChannel?: MessageChannel
   /** @inheritdoc */
   public tasksQueueBackPressureSize: number
-  /** @inheritdoc */
-  public onBackPressure?: WorkerNodeEventCallback
-  /** @inheritdoc */
-  public onEmptyQueue?: WorkerNodeEventCallback
   private readonly tasksQueue: Deque<Task<Data>>
   private onBackPressureStarted: boolean
   private onEmptyQueueCount: number
@@ -58,6 +55,7 @@ implements IWorkerNode<Worker, Data> {
    * @param tasksQueueBackPressureSize - The tasks queue back pressure size.
    */
   constructor (worker: Worker, tasksQueueBackPressureSize: number) {
+    super()
     checkWorkerNodeArguments<Worker>(worker, tasksQueueBackPressureSize)
     this.worker = worker
     this.info = this.initWorkerInfo(worker)
@@ -80,13 +78,13 @@ implements IWorkerNode<Worker, Data> {
   /** @inheritdoc */
   public enqueueTask (task: Task<Data>): number {
     const tasksQueueSize = this.tasksQueue.push(task)
-    if (
-      this.onBackPressure != null &&
-      this.hasBackPressure() &&
-      !this.onBackPressureStarted
-    ) {
+    if (this.hasBackPressure() && !this.onBackPressureStarted) {
       this.onBackPressureStarted = true
-      this.onBackPressure(this.info.id as number)
+      this.dispatchEvent(
+        new CustomEvent<WorkerNodeEventDetail>('backpressure', {
+          detail: { workerId: this.info.id as number }
+        })
+      )
       this.onBackPressureStarted = false
     }
     return tasksQueueSize
@@ -95,13 +93,13 @@ implements IWorkerNode<Worker, Data> {
   /** @inheritdoc */
   public unshiftTask (task: Task<Data>): number {
     const tasksQueueSize = this.tasksQueue.unshift(task)
-    if (
-      this.onBackPressure != null &&
-      this.hasBackPressure() &&
-      !this.onBackPressureStarted
-    ) {
+    if (this.hasBackPressure() && !this.onBackPressureStarted) {
       this.onBackPressureStarted = true
-      this.onBackPressure(this.info.id as number)
+      this.dispatchEvent(
+        new CustomEvent<WorkerNodeEventDetail>('backpressure', {
+          detail: { workerId: this.info.id as number }
+        })
+      )
       this.onBackPressureStarted = false
     }
     return tasksQueueSize
@@ -110,11 +108,7 @@ implements IWorkerNode<Worker, Data> {
   /** @inheritdoc */
   public dequeueTask (): Task<Data> | undefined {
     const task = this.tasksQueue.shift()
-    if (
-      this.onEmptyQueue != null &&
-      this.tasksQueue.size === 0 &&
-      this.onEmptyQueueCount === 0
-    ) {
+    if (this.tasksQueue.size === 0 && this.onEmptyQueueCount === 0) {
       this.startOnEmptyQueue().catch(EMPTY_FUNCTION)
     }
     return task
@@ -123,11 +117,7 @@ implements IWorkerNode<Worker, Data> {
   /** @inheritdoc */
   public popTask (): Task<Data> | undefined {
     const task = this.tasksQueue.pop()
-    if (
-      this.onEmptyQueue != null &&
-      this.tasksQueue.size === 0 &&
-      this.onEmptyQueueCount === 0
-    ) {
+    if (this.tasksQueue.size === 0 && this.onEmptyQueueCount === 0) {
       this.startOnEmptyQueue().catch(EMPTY_FUNCTION)
     }
     return task
@@ -152,10 +142,10 @@ implements IWorkerNode<Worker, Data> {
   /** @inheritdoc */
   public closeChannel (): void {
     if (this.messageChannel != null) {
-      this.messageChannel?.port1.unref()
-      this.messageChannel?.port2.unref()
-      this.messageChannel?.port1.close()
-      this.messageChannel?.port2.close()
+      this.messageChannel.port1.unref()
+      this.messageChannel.port2.unref()
+      this.messageChannel.port1.close()
+      this.messageChannel.port2.close()
       delete this.messageChannel
     }
   }
@@ -198,7 +188,11 @@ implements IWorkerNode<Worker, Data> {
       return
     }
     ++this.onEmptyQueueCount
-    this.onEmptyQueue?.(this.info.id as number)
+    this.dispatchEvent(
+      new CustomEvent<WorkerNodeEventDetail>('emptyqueue', {
+        detail: { workerId: this.info.id as number }
+      })
+    )
     await sleep(exponentialDelay(this.onEmptyQueueCount))
     await this.startOnEmptyQueue()
   }