refactoring

winglang · Apr 6, 2024 · 3e0e736 · 3e0e736
1 parent ccad721
commit 3e0e736
Show file tree

Hide file tree

Showing 4 changed files with 128 additions and 153 deletions.
diff --git a/dynamodb/dynamodb.extern.d.ts b/dynamodb/dynamodb.extern.d.ts
@@ -2,7 +2,7 @@ export default interface extern {
   createClient: (options: ClientConfig) => Promise<Client$Inflight>,
   createDocumentClient: (options?: (CreateDocumentClientOptions) | undefined) => Promise<DocumentClient$Inflight>,
   getPort: () => Promise<number>,
-  processRecordsAsync: (endpoint: string, tableName: string, handler: (arg0: StreamRecord) => Promise<void>, options?: (StreamConsumerOptions) | undefined) => Promise<void>,
+  processRecordsAsync: (endpoint: string, tableName: string, handler: (arg0: StreamRecord) => Promise<void>, eventName: string, options?: (StreamConsumerOptions) | undefined) => Promise<void>,
   unmarshall: (item: Readonly<any>, options?: (Readonly<any>) | undefined) => Promise<Readonly<any>>,
 }
 export interface Credentials {

diff --git a/dynamodb/dynamodb.mjs b/dynamodb/dynamodb.mjs
@@ -48,23 +48,25 @@ const processStreamRecords = async (client, StreamArn, handler, eventType, optio
       });
 
       for (const record of recordsData.Records) {
-        try {
-          await handler({
-            eventId: record.eventID,
-            eventName: record.eventName,
-            dynamodb: {
-              ApproximateCreationDateTime:
-                record.dynamodb.ApproximateCreationDateTime,
-              Keys: record.dynamodb.Keys,
-              NewImage: record.dynamodb.NewImage,
-              OldImage: record.dynamodb.OldImage,
-              SequenceNumber: record.dynamodb.SequenceNumber,
-              SizeBytes: record.dynamodb.SizeBytes,
-              StreamViewType: record.dynamodb.StreamViewType,
-            },
-          });
-        } catch (error) {
-          console.error("Error processing stream record:", error, record);
+        if (!event || record.eventName == event) {
+          try {
+            await handler({
+              eventId: record.eventID,
+              eventName: record.eventName,
+              dynamodb: {
+                ApproximateCreationDateTime:
+                  record.dynamodb.ApproximateCreationDateTime,
+                Keys: record.dynamodb.Keys,
+                NewImage: record.dynamodb.NewImage,
+                OldImage: record.dynamodb.OldImage,
+                SequenceNumber: record.dynamodb.SequenceNumber,
+                SizeBytes: record.dynamodb.SizeBytes,
+                StreamViewType: record.dynamodb.StreamViewType,
+              },
+            });
+          } catch (error) {
+            console.error("Error processing stream record:", error, record);
+          }
         }
       }