AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Parameters:
  MyBatchingWindowParam:
    Type: Number
    Default: 45
    Description: parameter for batching window in seconds

Resources:
  MyFunctionForBatchingExample:
    Type: AWS::Serverless::Function
    Properties:
      Handler: index.handler
      InlineCode: |
        exports.handler = async (event) => {
            return {
            statusCode: 200,
            body: JSON.stringify(event),
            headers: {}
            }
        }
      Runtime: nodejs12.x
      Policies:
      - SQSSendMessagePolicy:
          QueueName: !GetAtt MySqsQueue.QueueName
      Events:
        Stream:
          Type: Kinesis
          Properties:
            Stream: !GetAtt KinesisStream.Arn
            MaximumBatchingWindowInSeconds: 20
            StartingPosition: LATEST
        StreamEvent:
          Type: Kinesis
          Properties:
            Stream: !GetAtt KinesisStream1.Arn
            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
            StartingPosition: LATEST
            DestinationConfig:
              OnFailure:
                Type: SNS
                Destination: !Ref MySnsTopic
        DynamoDBStreamEvent:
          Type: DynamoDB
          Properties:
            Stream: !GetAtt DynamoDBTable.StreamArn
            BatchSize: 100
            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
            ParallelizationFactor: 8
            MaximumRetryAttempts: 100
            BisectBatchOnFunctionError: true
            MaximumRecordAgeInSeconds: 86400
            StartingPosition: TRIM_HORIZON
            TumblingWindowInSeconds: 60
            FunctionResponseTypes:
            - ReportBatchItemFailures
            DestinationConfig:
              OnFailure:
                Type: SQS
                Destination: !GetAtt MySqsQueue.Arn
        StreamEventWithoutDestinationConfigType:
          Type: Kinesis
          Properties:
            Stream: !GetAtt KinesisStream1.Arn
            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
            StartingPosition: LATEST
            DestinationConfig:
              OnFailure:
                Destination: !Ref MySnsTopic
        StreamEventWithEmptyDestinationConfig:
          Type: Kinesis
          Properties:
            Stream: !GetAtt KinesisStream1.Arn
            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
            StartingPosition: LATEST
            DestinationConfig:

  KinesisStream:
    Type: AWS::Kinesis::Stream
    Properties:
      ShardCount: 1

  KinesisStream1:
    Type: AWS::Kinesis::Stream
    Properties:
      ShardCount: 1
  DynamoDBTable:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
      - AttributeName: id
        AttributeType: S
      KeySchema:
      - AttributeName: id
        KeyType: HASH
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5
      StreamSpecification:
        StreamViewType: NEW_IMAGE

  MySqsQueue:
    Type: AWS::SQS::Queue
  MySnsTopic:
    Type: AWS::SNS::Topic