perf: lambda async sources

brexhq · Nov 1, 2023 · 49d5c4b · 49d5c4b
1 parent c73e2b8
commit 49d5c4b
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 48 deletions.
diff --git a/cmd/aws/lambda/substation/dynamodb.go b/cmd/aws/lambda/substation/dynamodb.go
@@ -60,17 +60,12 @@ func dynamodbHandler(ctx context.Context, event events.DynamoDBEvent) error {
 
 			m := message
 			tfGroup.Go(func() error {
-				msg, err := transform.Apply(tfCtx, sub.Transforms(), m)
-				if err != nil {
+				// Transformed messages are never returned to the caller because
+				// invocation is asynchronous.
+				if _, err := transform.Apply(tfCtx, sub.Transforms(), m); err != nil {
 					return err
 				}
 
-				for _, m := range msg {
-					if m.IsControl() {
-						continue
-					}
-				}
-
 				return nil
 			})
 		}

diff --git a/cmd/aws/lambda/substation/kinesis_stream.go b/cmd/aws/lambda/substation/kinesis_stream.go
@@ -57,17 +57,12 @@ func kinesisStreamHandler(ctx context.Context, event events.KinesisEvent) error
 
 			m := message
 			tfGroup.Go(func() error {
-				msg, err := transform.Apply(tfCtx, sub.Transforms(), m)
-				if err != nil {
+				// Transformed messages are never returned to the caller because
+				// invocation is asynchronous.
+				if _, err := transform.Apply(tfCtx, sub.Transforms(), m); err != nil {
 					return err
 				}
 
-				for _, m := range msg {
-					if m.IsControl() {
-						continue
-					}
-				}
-
 				return nil
 			})
 		}

diff --git a/cmd/aws/lambda/substation/s3.go b/cmd/aws/lambda/substation/s3.go
@@ -62,17 +62,12 @@ func s3Handler(ctx context.Context, event events.S3Event) error {
 
 			m := message
 			tfGroup.Go(func() error {
-				msg, err := transform.Apply(tfCtx, sub.Transforms(), m)
-				if err != nil {
+				// Transformed messages are never returned to the caller because
+				// invocation is asynchronous.
+				if _, err := transform.Apply(tfCtx, sub.Transforms(), m); err != nil {
 					return err
 				}
 
-				for _, m := range msg {
-					if m.IsControl() {
-						continue
-					}
-				}
-
 				return nil
 			})
 		}
@@ -206,17 +201,12 @@ func s3SnsHandler(ctx context.Context, event events.SNSEvent) error {
 
 			m := message
 			tfGroup.Go(func() error {
-				msg, err := transform.Apply(tfCtx, sub.Transforms(), m)
-				if err != nil {
+				// Transformed messages are never returned to the caller because
+				// invocation is asynchronous.
+				if _, err := transform.Apply(tfCtx, sub.Transforms(), m); err != nil {
 					return err
 				}
 
-				for _, m := range msg {
-					if m.IsControl() {
-						continue
-					}
-				}
-
 				return nil
 			})
 		}

diff --git a/cmd/aws/lambda/substation/sns.go b/cmd/aws/lambda/substation/sns.go
@@ -56,17 +56,12 @@ func snsHandler(ctx context.Context, event events.SNSEvent) error {
 
 			m := message
 			tfGroup.Go(func() error {
-				msg, err := transform.Apply(tfCtx, sub.Transforms(), m)
-				if err != nil {
+				// Transformed messages are never returned to the caller because
+				// invocation is asynchronous.
+				if _, err := transform.Apply(tfCtx, sub.Transforms(), m); err != nil {
 					return err
 				}
 
-				for _, m := range msg {
-					if m.IsControl() {
-						continue
-					}
-				}
-
 				return nil
 			})
 		}

diff --git a/cmd/aws/lambda/substation/sqs.go b/cmd/aws/lambda/substation/sqs.go
@@ -55,17 +55,12 @@ func sqsHandler(ctx context.Context, event events.SQSEvent) error {
 
 			m := message
 			tfGroup.Go(func() error {
-				msg, err := transform.Apply(tfCtx, sub.Transforms(), m)
-				if err != nil {
+				// Transformed messages are never returned to the caller because
+				// invocation is asynchronous.
+				if _, err := transform.Apply(tfCtx, sub.Transforms(), m); err != nil {
 					return err
 				}
 
-				for _, m := range msg {
-					if m.IsControl() {
-						continue
-					}
-				}
-
 				return nil
 			})
 		}