Homework/cs5260/consumer/Loop.ts


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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})...`);
};