+ this.sendToWorker(workerNodeKey, message)
+ })
+ }
+
+ private async sendTaskFunctionOperationToWorkers (
+ message: MessageValue<Data>
+ ): Promise<boolean> {
+ return await new Promise<boolean>((resolve, reject) => {
+ const responsesReceived = new Array<MessageValue<Response>>()
+ const taskFunctionOperationsListener = (
+ message: MessageValue<Response>
+ ): void => {
+ this.checkMessageWorkerId(message)
+ if (message.taskFunctionOperationStatus != null) {
+ responsesReceived.push(message)
+ if (responsesReceived.length === this.workerNodes.length) {
+ if (
+ responsesReceived.every(
+ message => message.taskFunctionOperationStatus === true
+ )
+ ) {
+ resolve(true)
+ } else if (
+ responsesReceived.some(
+ message => message.taskFunctionOperationStatus === false
+ )
+ ) {
+ const errorResponse = responsesReceived.find(
+ response => response.taskFunctionOperationStatus === false
+ )
+ reject(
+ new Error(
+ `Task function operation '${
+ message.taskFunctionOperation as string
+ }' failed on worker ${
+ errorResponse?.workerId as number
+ } with error: '${
+ errorResponse?.workerError?.message as string
+ }'`
+ )
+ )
+ }
+ this.deregisterWorkerMessageListener(
+ this.getWorkerNodeKeyByWorkerId(message.workerId),
+ taskFunctionOperationsListener
+ )
+ }
+ }
+ }
+ for (const [workerNodeKey] of this.workerNodes.entries()) {
+ this.registerWorkerMessageListener(
+ workerNodeKey,
+ taskFunctionOperationsListener
+ )
+ this.sendToWorker(workerNodeKey, message)
+ }
+ })
+ }
+
+ /** @inheritDoc */
+ public hasTaskFunction (name: string): boolean {
+ for (const workerNode of this.workerNodes) {
+ if (
+ Array.isArray(workerNode.info.taskFunctionNames) &&
+ workerNode.info.taskFunctionNames.includes(name)
+ ) {
+ return true
+ }
+ }
+ return false
+ }
+
+ /** @inheritDoc */
+ public async addTaskFunction (
+ name: string,
+ fn: TaskFunction<Data, Response>
+ ): Promise<boolean> {
+ if (typeof name !== 'string') {
+ throw new TypeError('name argument must be a string')
+ }
+ if (typeof name === 'string' && name.trim().length === 0) {
+ throw new TypeError('name argument must not be an empty string')
+ }
+ if (typeof fn !== 'function') {
+ throw new TypeError('fn argument must be a function')
+ }
+ const opResult = await this.sendTaskFunctionOperationToWorkers({
+ taskFunctionOperation: 'add',
+ taskFunctionName: name,
+ taskFunction: fn.toString()
+ })
+ this.taskFunctions.set(name, fn)
+ return opResult
+ }
+
+ /** @inheritDoc */
+ public async removeTaskFunction (name: string): Promise<boolean> {
+ if (!this.taskFunctions.has(name)) {
+ throw new Error(
+ 'Cannot remove a task function not handled on the pool side'
+ )
+ }
+ const opResult = await this.sendTaskFunctionOperationToWorkers({
+ taskFunctionOperation: 'remove',
+ taskFunctionName: name
+ })
+ this.deleteTaskFunctionWorkerUsages(name)
+ this.taskFunctions.delete(name)
+ return opResult
+ }
+
+ /** @inheritDoc */
+ public listTaskFunctionNames (): string[] {
+ for (const workerNode of this.workerNodes) {
+ if (
+ Array.isArray(workerNode.info.taskFunctionNames) &&
+ workerNode.info.taskFunctionNames.length > 0
+ ) {
+ return workerNode.info.taskFunctionNames
+ }