Fix memory leaking

nodejs/node#6673

src/command.ts

@@ -98,16 +98,17 @@ const processCommands = async (
 };
 
 export const executor = async () => {
-  try {
-    const commands: Command.AllCommand[] = await poll(commandConsumerClient);
-    if (commands.length) {
-      await processCommands(commands);
-      commandConsumerClient.commit();
+  while (true) {
+    try {
+      const commands: Command.AllCommand[] = await poll(commandConsumerClient);
+      if (commands.length) {
+        await processCommands(commands);
+        commandConsumerClient.commit();
+      }
+    } catch (error) {
+      // Handle consume error
+      console.warn('command', error);
+      await sleep(1000);
     }
-  } catch (error) {
-    // Handle consume error
-    console.warn(error);
-    await sleep(1000);
   }
-  setImmediate(executor);
 };

src/state.ts

@@ -716,30 +716,30 @@ export const processUpdateTasks = async (
 };
 
 export const executor = async () => {
-  try {
-    const tasksUpdate: Event.ITaskUpdate[] = await poll(
-      stateConsumerClient,
-      200,
-    );
-    if (tasksUpdate.length) {
-      // Grouped by workflowId, so it can execute parallely, but same workflowId have to run sequential bacause it can effect each other
-      const groupedTasks = R.values(
-        R.groupBy(R.path(['workflowId']), tasksUpdate),
-      );
-
-      await Promise.all(
-        groupedTasks.map((workflowTasksUpdate: Event.ITaskUpdate[]) =>
-          processUpdateTasks(workflowTasksUpdate),
-        ),
+  while (true) {
+    try {
+      const tasksUpdate: Event.ITaskUpdate[] = await poll(
+        stateConsumerClient,
+        200,
       );
-
-      stateConsumerClient.commit();
+      if (tasksUpdate.length) {
+        // Grouped by workflowId, so it can execute parallely, but same workflowId have to run sequential bacause it can effect each other
+        const groupedTasks = R.values(
+          R.groupBy(R.path(['workflowId']), tasksUpdate),
+        );
+
+        await Promise.all(
+          groupedTasks.map((workflowTasksUpdate: Event.ITaskUpdate[]) =>
+            processUpdateTasks(workflowTasksUpdate),
+          ),
+        );
+
+        stateConsumerClient.commit();
+      }
+    } catch (error) {
+      // Handle error here
+      console.warn('state', error);
+      await sleep(1000);
     }
-  } catch (error) {
-    // Handle error here
-    console.warn(error);
-    await sleep(1000);
-  } finally {
-    setImmediate(executor);
   }
 };