1 files changed, 47 insertions, 0 deletions
diff --git a/Homework/cs5260/consumer/Loop.ts b/Homework/cs5260/consumer/Loop.ts
new file mode 100644
index 0000000..e8e88d2
--- /dev/null
+++ b/Homework/cs5260/consumer/Loop.ts
@@ -0,0 +1,47 @@
+import { TracingLogger } from "./log";
+import { WidgetRequestProcessor } from "./processor";
+import { WidgetRequestRetriever } from "./retrievers";
+import { WidgetRequestT } from "./schema";
+
+const sleepFor = (ms: number) => new Promise((res) => setTimeout(res, ms));
+
+export const MainLoop = async (
+  retrievers: WidgetRequestRetriever[],
+  processors: WidgetRequestProcessor[],
+  maxPollAttempts: number,
+  pollIntervalMs: number,
+  logger: TracingLogger,
+) => {
+  let pollAttemptsWithNoResults = 0;
+  while (pollAttemptsWithNoResults < maxPollAttempts) {
+    logger.info(`polling...`);
+
+    let requests: WidgetRequestT[] = [];
+    for (const retriever of retrievers) {
+      const retrievalRequests = await retriever.refreshRequests();
+      requests = requests.concat(retrievalRequests);
+    }
+    logger.info(`received Requests=(${requests.length}) requests...`);
+
+    for (const request of requests) {
+      for (const processor of processors) {
+        await processor.processRequest(request);
+      }
+    }
+    logger.info(
+      `finished processing Requests=(${requests.length}) requests...`,
+    );
+
+    if (requests.length === 0) {
+      pollAttemptsWithNoResults++;
+      logger.info(
+        `sleeping for Interval=(${pollIntervalMs})ms: (${pollAttemptsWithNoResults} / ${maxPollAttempts})`,
+      );
+      await sleepFor(pollIntervalMs);
+    } else {
+      pollAttemptsWithNoResults = 0;
+    }
+  }
+
+  logger.info(`no requests after Attempts=(${maxPollAttempts})...`);
+};