Throttle the live streaming of the run to 1 second updates (triggerdotdev#972)

matt-aitken · web-flow · commit 9c89ab40aa0a · 2024-03-26T14:20:10.000Z
diff --git a/apps/webapp/app/presenters/v3/RunStreamPresenter.server.ts b/apps/webapp/app/presenters/v3/RunStreamPresenter.server.ts
@@ -1,17 +1,10 @@
-import { TaskRun, TaskRunAttempt } from "@trigger.dev/database";
+import { TaskRun } from "@trigger.dev/database";
 import { eventStream } from "remix-utils/sse/server";
 import { PrismaClient, prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
+import { throttle } from "~/utils/throttle";
 import { eventRepository } from "~/v3/eventRepository.server";
 
-type RunWithAttempts = {
-  updatedAt: Date;
-  attempts: {
-    status: TaskRunAttempt["status"];
-    updatedAt: Date;
-  }[];
-};
-
 const pingInterval = 1000;
 
 export class RunStreamPresenter {
@@ -77,8 +70,10 @@ export class RunStreamPresenter {
         }
       };
 
+      const throttledSend = throttle(safeSend, 1000);
+
       eventEmitter.addListener("message", (event) => {
-        safeSend({ data: event });
+        throttledSend({ data: event });
       });
 
       pinger = setInterval(() => {
diff --git a/apps/webapp/app/utils/throttle.ts b/apps/webapp/app/utils/throttle.ts
@@ -0,0 +1,20 @@
+//From: https://kettanaito.com/blog/debounce-vs-throttle
+
+/** A very simple throttle. Will execute the function every Xms and discard any other calls during that period. */
+export function throttle(
+  func: (...args: any[]) => void,
+  duration: number
+): (...args: any[]) => void {
+  let shouldWait = false;
+
+  return (...args: any[]) => {
+    if (!shouldWait) {
+      func(...args);
+      shouldWait = true;
+
+      setTimeout(() => {
+        shouldWait = false;
+      }, duration);
+    }
+  };
+}
