1 files changed, 60 insertions, 0 deletions
diff --git a/Homework/cs5260/consumer/retrievers/SQSWidgetRequestRetriever.ts b/Homework/cs5260/consumer/retrievers/SQSWidgetRequestRetriever.ts
new file mode 100644
index 0000000..40a837b
--- /dev/null
+++ b/Homework/cs5260/consumer/retrievers/SQSWidgetRequestRetriever.ts
@@ -0,0 +1,60 @@
+import {
+  DeleteMessageBatchCommand,
+  ReceiveMessageCommand,
+  SQSClient,
+} from "@aws-sdk/client-sqs";
+import { WidgetRequestRetriever } from ".";
+import { TracingLogger } from "@/shared/log";
+import { WidgetRequest, WidgetRequestT } from "@/shared/schema";
+
+export class SQSWidgetRequestRetriever implements WidgetRequestRetriever {
+  static MESSAGES_PER_POLL = 10;
+  private queueUrl: string;
+  private sqsClient: SQSClient;
+  private logger: TracingLogger;
+
+  constructor(queueUrl: string, sqsClient: SQSClient, logger: TracingLogger) {
+    this.queueUrl = queueUrl;
+    this.sqsClient = sqsClient;
+    this.logger = logger;
+  }
+
+  public async refreshRequests() {
+    const retrieveMessagesRequest = new ReceiveMessageCommand({
+      QueueUrl: this.queueUrl,
+      MaxNumberOfMessages: SQSWidgetRequestRetriever.MESSAGES_PER_POLL,
+    });
+
+    this.logger.info(
+      `looking for Messages=(${SQSWidgetRequestRetriever.MESSAGES_PER_POLL}) WidgetRequest's from Queue=(${this.queueUrl})`,
+    );
+    const response = await this.sqsClient.send(retrieveMessagesRequest);
+    if (!response.Messages) {
+      this.logger.info(`Queue=(${this.queueUrl}) returned no messages!`);
+      return [];
+    }
+
+    const widgetRequests = response.Messages.map((message) => {
+      if (message.Body) return WidgetRequest.parse(JSON.parse(message.Body));
+    }).filter((x) => x) as WidgetRequestT[];
+    this.logger.info(
+      `found WidgetRequests=(${widgetRequests.length}) requests`,
+    );
+
+    const messageDeleteInputs = response.Messages.map(
+      ({ ReceiptHandle, MessageId }) => {
+        return { Id: MessageId, ReceiptHandle };
+      },
+    );
+    this.logger.info(
+      `requesting deletion of Inputs=(${messageDeleteInputs.length})`,
+    );
+    const deleteRequestCommand = new DeleteMessageBatchCommand({
+      QueueUrl: this.queueUrl,
+      Entries: messageDeleteInputs,
+    });
+    await this.sqsClient.send(deleteRequestCommand);
+
+    return widgetRequests;
+  }
+}