feat: Stream Analytics

Author: mndeveci

samtranslator/model/eventsources/pull.py

@@ -34,6 +34,7 @@ class PullEventSource(ResourceMacro):
         "Broker": PropertyType(False, is_str()),
         "Queues": PropertyType(False, is_type(list)),
         "SourceAccessConfigurations": PropertyType(False, is_type(list)),
+        "TumblingWindowInSeconds": PropertyType(False, is_type(int)),
     }
 
     def get_policy_arn(self):
@@ -85,6 +86,7 @@ def to_cloudformation(self, **kwargs):
         lambda_eventsourcemapping.Topics = self.Topics
         lambda_eventsourcemapping.Queues = self.Queues
         lambda_eventsourcemapping.SourceAccessConfigurations = self.SourceAccessConfigurations
+        lambda_eventsourcemapping.TumblingWindowInSeconds = self.TumblingWindowInSeconds
 
         destination_config_policy = None
         if self.DestinationConfig:

samtranslator/model/lambda_.py

@@ -75,6 +75,7 @@ class LambdaEventSourceMapping(Resource):
         "Topics": PropertyType(False, is_type(list)),
         "Queues": PropertyType(False, is_type(list)),
         "SourceAccessConfigurations": PropertyType(False, is_type(list)),
+        "TumblingWindowInSeconds": PropertyType(False, is_type(int)),
     }
 
     runtime_attrs = {"name": lambda self: ref(self.logical_id)}

tests/translator/input/function_with_event_source_mapping.yaml

@@ -52,6 +52,7 @@ Resources:
             BisectBatchOnFunctionError: true
             MaximumRecordAgeInSeconds: 86400
             StartingPosition: TRIM_HORIZON
+            TumblingWindowInSeconds: 60
             DestinationConfig:
               OnFailure:
                 Type: SQS

tests/translator/output/aws-cn/function_with_event_source_mapping.json

@@ -181,7 +181,8 @@
         "StartingPosition": "TRIM_HORIZON",
         "ParallelizationFactor": 8,
         "MaximumRetryAttempts": 100,
-        "BisectBatchOnFunctionError": true
+        "BisectBatchOnFunctionError": true,
+        "TumblingWindowInSeconds": 60
       }
     },
     "MyFunctionForBatchingExample": {

tests/translator/output/aws-us-gov/function_with_event_source_mapping.json

@@ -181,7 +181,8 @@
         "StartingPosition": "TRIM_HORIZON",
         "ParallelizationFactor": 8,
         "MaximumRetryAttempts": 100,
-        "BisectBatchOnFunctionError": true
+        "BisectBatchOnFunctionError": true,
+        "TumblingWindowInSeconds": 60
       }
     },
     "MyFunctionForBatchingExample": {

tests/translator/output/function_with_event_source_mapping.json

@@ -181,7 +181,8 @@
         "StartingPosition": "TRIM_HORIZON",
         "ParallelizationFactor": 8,
         "MaximumRetryAttempts": 100,
-        "BisectBatchOnFunctionError": true
+        "BisectBatchOnFunctionError": true,
+        "TumblingWindowInSeconds": 60
       }
     },
     "MyFunctionForBatchingExample": {

versions/2016-10-31.md

@@ -512,6 +512,7 @@ BisectBatchOnFunctionError | `boolean` | A boolean flag which determines whether
 MaximumRecordAgeInSeconds | `integer` | The maximum age of a record that will be invoked by Lambda. If an `OnFailure` destination is set, metadata describing the records will be sent to the destination. If no destination is set, the records will be bypassed
 DestinationConfig | [Destination Config Object](#destination-config-object) | Expired record metadata/retries and exhausted metadata is sent to this destination after they have passed the defined limits.
 ParallelizationFactor | `integer` | Allocates multiple virtual shards, increasing the Lambda invokes by the given factor and speeding up the stream processing.
+TumblingWindowInSeconds | `integer` |  Tumbling window (non-overlapping time window) duration to perform aggregations.
 
 **NOTE:** `SQSSendMessagePolicy` or `SNSPublishMessagePolicy` needs to be added in `Policies` for publishing messages to the `SQS` or `SNS` resource mentioned in `OnFailure` property
 
@@ -534,6 +535,7 @@ Properties:
     OnFailure:
       Type: SQS
       Destination: !GetAtt MySqsQueue.Arn
+  TumblingWindowInSeconds: 0
 ```
 
 
@@ -579,6 +581,7 @@ BisectBatchOnFunctionError | `boolean` | A boolean flag which determines whether
 MaximumRecordAgeInSeconds | `integer` | The maximum age of a record that will be invoked by Lambda. If an `OnFailure` destination is set, metadata describing the records will be sent to the destination. If no destination is set, the records will be bypassed
 DestinationConfig | [DestinationConfig Object](#destination-config-object) | Expired record metadata/retries and exhausted metadata is sent to this destination after they have passed the defined limits.
 ParallelizationFactor | `integer` | Allocates multiple virtual shards, increasing the Lambda invokes by the given factor and speeding up the stream processing.
+TumblingWindowInSeconds | `integer` |  Tumbling window (non-overlapping time window) duration to perform aggregations.
 
 ##### Example: DynamoDB event source object
 
@@ -598,6 +601,7 @@ Properties:
     OnFailure:
       Type: SQS
       Destination: !GetAtt MySqsQueue.Arn
+  TumblingWindowInSeconds: 0
 ```
 
 #### SQS