fix(bpmn-service): fix worker execution issue in bpmn service

services/bpmn-service/src/controllers/workflow.controller.ts

@@ -264,7 +264,7 @@ export class WorkflowController {
       const workerList = workerMap[workflowName];
       for (const worker of workerList) {
         if (!worker.running) {
-          await this.workerFn(worker.topic, worker.command);
+          await this.workerFn(worker);
           worker.running = true;
         }
       }

services/bpmn-service/src/providers/worker-implementation.provider.ts

@@ -25,14 +25,25 @@ export class WorkerImplementationProvider
     }
   }
   value(): WorkerImplementationFn {
-    return async (topic, cmd) => {
+    return async worker => {
       if (this.client) {
-        this.client.subscribe(topic, ({task, taskService}) => {
-          cmd.operation({task, taskService}, (result: AnyObject) => {
-            if (result) {
-              this.ilogger.info(`Worker task completed - ${topic}`);
-            }
-          });
+        worker.running = true;
+        const subscription = this.client.subscribe(
+          worker.topic,
+          ({task, taskService}) => {
+            worker.command.operation(
+              {task, taskService},
+              (result: AnyObject) => {
+                if (result) {
+                  this.ilogger.info(`Worker task completed - ${worker.topic}`);
+                }
+              },
+            );
+          },
+        );
+        this.client.on('poll:error', () => {
+          worker.running = false;
+          subscription.unsubscribe();
         });
       } else {
         throw new Error('Workflow client not connected');

services/bpmn-service/src/types.ts

@@ -55,7 +55,7 @@ export type WorkerNameCmdPair<T = AnyObject, R = AnyObject> = {
 };
 
 export interface WorkerImplementationFn<T = AnyObject, R = AnyObject> {
-  (topicName: string, commandCtor: BPMTask<T, R>): Promise<void>;
+  (worker: WorkerNameCmdPair<T, R>): Promise<void>;
 }
 
 export type SuccessResponse = {