From 6b272951b296a2ea26659e9a2fa81bdc6003407f Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 2 Jul 2023 22:53:31 +0200 Subject: [PATCH] refactor: factor out worker message handlers MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- src/pools/abstract-pool.ts | 75 ++++++++++++++++++++++---------------- 1 file changed, 43 insertions(+), 32 deletions(-) diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 24e8a5b0..6b28833d 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -816,42 +816,53 @@ export abstract class AbstractPool< return message => { if (message.workerId != null && message.started != null) { // Worker started message received - const worker = this.getWorkerById(message.workerId) - if (worker != null) { - this.workerNodes[this.getWorkerNodeKey(worker)].info.started = - message.started - } else { - throw new Error( - `Worker started message received from unknown worker '${message.workerId}'` - ) - } + this.handleWorkerStartedMessage(message) } else if (message.id != null) { // Task execution response received - const promiseResponse = this.promiseResponseMap.get(message.id) - if (promiseResponse != null) { - if (message.taskError != null) { - if (this.emitter != null) { - this.emitter.emit(PoolEvents.taskError, message.taskError) - } - promiseResponse.reject(message.taskError.message) - } else { - promiseResponse.resolve(message.data as Response) - } - this.afterTaskExecutionHook(promiseResponse.worker, message) - this.promiseResponseMap.delete(message.id) - const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker) - if ( - this.opts.enableTasksQueue === true && - this.tasksQueueSize(workerNodeKey) > 0 - ) { - this.executeTask( - workerNodeKey, - this.dequeueTask(workerNodeKey) as Task - ) - } - this.workerChoiceStrategyContext.update(workerNodeKey) + this.handleTaskExecutionResponse(message) + } + } + } + + private handleWorkerStartedMessage (message: MessageValue): void { + // Worker started message received + const worker = this.getWorkerById(message.workerId as number) + if (worker != null) { + this.workerNodes[this.getWorkerNodeKey(worker)].info.started = + message.started as boolean + } else { + throw new Error( + `Worker started message received from unknown worker '${ + message.workerId as number + }'` + ) + } + } + + private handleTaskExecutionResponse (message: MessageValue): void { + const promiseResponse = this.promiseResponseMap.get(message.id as string) + if (promiseResponse != null) { + if (message.taskError != null) { + if (this.emitter != null) { + this.emitter.emit(PoolEvents.taskError, message.taskError) } + promiseResponse.reject(message.taskError.message) + } else { + promiseResponse.resolve(message.data as Response) + } + this.afterTaskExecutionHook(promiseResponse.worker, message) + this.promiseResponseMap.delete(message.id as string) + const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker) + if ( + this.opts.enableTasksQueue === true && + this.tasksQueueSize(workerNodeKey) > 0 + ) { + this.executeTask( + workerNodeKey, + this.dequeueTask(workerNodeKey) as Task + ) } + this.workerChoiceStrategyContext.update(workerNodeKey) } } -- 2.34.1