InitHEADmain

1 files changed, 113 insertions, 0 deletions

diff --git a/Homework/cs5260/consumer/index.ts b/Homework/cs5260/consumer/index.ts
new file mode 100644
index 0000000..f8ce640
--- /dev/null
+++ b/Homework/cs5260/consumer/index.ts
@@ -0,0 +1,113 @@
+import { S3Client } from "@aws-sdk/client-s3";
+import { SQSClient } from "@aws-sdk/client-sqs";
+import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
+
+import { Args, args } from "@/consumer/args";
+import { DynamoWidgetDAO, S3WidgetDAO } from "@/consumer/dao";
+import { S3WidgetKeyGenerator } from "@/consumer/dao/keys";
+import { ConsoleTracingLogger } from "@/shared/log";
+import { ConsumerRequestProcessor, WidgetRequestProcessor } from "@/consumer/processor";
+import {
+  S3WidgetRequestRetriever,
+  WidgetRequestRetriever,
+  SQSWidgetRequestRetriever,
+} from "@/consumer/retrievers";
+import { MainLoop } from "@/consumer/Loop";
+
+const main = async (args: Args) => {
+  const mainLogger = new ConsoleTracingLogger("main");
+
+  const s3Client = new S3Client();
+  const sqsClient = new SQSClient();
+  const dynamoClient = new DynamoDBClient();
+
+  const retrievers: WidgetRequestRetriever[] = [];
+  if (!args.sourcebucket && !args.sourcequeue) {
+    throw new Error("Must specify source to poll.");
+  }
+
+  if (args.sourcebucket) {
+    mainLogger.info(
+      `constructing s3 retriever for Bucket=(${args.sourcebucket})`,
+    );
+
+    const retriever = new S3WidgetRequestRetriever(
+      args.sourcebucket,
+      s3Client,
+      mainLogger.createChild("S3WidgetRequestRetriever"),
+    );
+    retrievers.push(retriever);
+  }
+
+  if (args.sourcequeue) {
+    mainLogger.info(
+      `constructing sqs retriever for Bucket=(${args.sourcequeue})`,
+    );
+
+    const retriever = new SQSWidgetRequestRetriever(
+      args.sourcequeue,
+      sqsClient,
+      mainLogger.createChild("SQSWidgetRequestRetriever"),
+    );
+    retrievers.push(retriever);
+  }
+
+  if (!args.storagebucket && !args.storagetable) {
+    throw new Error("Must specify a target data store for Widgets");
+  }
+
+  const processors: WidgetRequestProcessor[] = [];
+
+  if (args.storagebucket) {
+    if (!args.storageprefix) {
+      throw new Error("must specify widget prefix location for storage bucket");
+    }
+
+    mainLogger.info(
+      `constructing s3 processor for Bucket=(${args.storagebucket})`,
+    );
+
+    const s3KeyGenerator = new S3WidgetKeyGenerator().withPrefix(
+      args.storageprefix,
+    );
+    const s3WidgetDao = new S3WidgetDAO(
+      args.storagebucket,
+      s3Client,
+      s3KeyGenerator,
+      mainLogger.createChild("S3WidgetDAO"),
+    );
+
+    const s3RequestProcessor = new ConsumerRequestProcessor(
+      s3WidgetDao,
+      mainLogger.createChild("ConsumerRequestProcessor => (S3)"),
+    );
+
+    processors.push(s3RequestProcessor);
+  }
+
+  if (args.storagetable) {
+    mainLogger.info(
+      `constructing dynamo processor for Table=(${args.storagetable})`,
+    );
+
+    const dynamoWidgetDao = new DynamoWidgetDAO(
+      args.storagetable,
+      dynamoClient,
+      mainLogger.createChild("DynamoWidgetDAO"),
+    );
+
+    const dynamoRequestProcessor = new ConsumerRequestProcessor(
+      dynamoWidgetDao,
+      mainLogger.createChild("ConsumerRequestProcessor => (Dynamo)"),
+    );
+
+    processors.push(dynamoRequestProcessor);
+  }
+
+  const maxPollAttempts = 15;
+  await MainLoop(retrievers, processors, maxPollAttempts, 200, mainLogger);
+
+  mainLogger.info("EXITING!");
+};
+
+main(args);