Multi dest connector

integration/combination/test_connectors.py

@@ -28,6 +28,7 @@ def tearDown(self):
             ("combination/connector_restapi_to_function",),
             ("combination/connector_httpapi_to_function",),
             ("combination/connector_function_to_bucket_read",),
+            ("combination/connector_function_to_bucket_read_multiple",),
             ("combination/connector_function_to_bucket_write",),
             ("combination/connector_function_to_table_read",),
             ("combination/connector_function_to_table_write",),
@@ -44,6 +45,8 @@ def tearDown(self):
             ("combination/connector_event_rule_to_eb_default_write",),
             ("combination/connector_event_rule_to_eb_custom_write",),
             ("combination/connector_event_rule_to_lambda_write",),
+            ("combination/connector_event_rule_to_lambda_write_multiple",),
+            ("combination/connector_mix_destination",),
             ("combination/connector_sqs_to_function",),
             ("combination/connector_sns_to_function_write",),
             ("combination/connector_table_to_function_read",),

integration/resources/expected/combination/connector_event_rule_to_lambda_write_multiple.json

@@ -0,0 +1,42 @@
+[
+  {
+    "LogicalResourceId": "EventRule",
+    "ResourceType": "AWS::Events::Rule"
+  },
+  {
+    "LogicalResourceId": "Function",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "FunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "Function2",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "Function2Role",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "MyConnectorWriteLambdaPermissionDestination0",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "MyConnectorWriteLambdaPermissionDestination1",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "TriggerFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "TriggerFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "VerificationQueue",
+    "ResourceType": "AWS::SQS::Queue"
+  }
+]

integration/resources/expected/combination/connector_function_to_bucket_read_multiple.json

@@ -0,0 +1,26 @@
+[
+  {
+    "LogicalResourceId": "LambdaRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "TriggerFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "Bucket",
+    "ResourceType": "AWS::S3::Bucket"
+  },
+  {
+    "LogicalResourceId": "Bucket2",
+    "ResourceType": "AWS::S3::Bucket"
+  },
+  {
+    "LogicalResourceId": "ConnectorPolicyDestination0",
+    "ResourceType": "AWS::IAM::ManagedPolicy"
+  },
+  {
+    "LogicalResourceId": "ConnectorPolicyDestination1",
+    "ResourceType": "AWS::IAM::ManagedPolicy"
+  }
+]

integration/resources/expected/combination/connector_mix_destination.json

@@ -0,0 +1,34 @@
+[
+  {
+    "LogicalResourceId": "EventRule",
+    "ResourceType": "AWS::Events::Rule"
+  },
+  {
+    "LogicalResourceId": "Function",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "FunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "MyConnectorWriteLambdaPermissionDestination0",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "TriggerFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "TriggerFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "VerificationQueue",
+    "ResourceType": "AWS::SQS::Queue"
+  },
+  {
+    "LogicalResourceId": "MyConnectorQueuePolicyDestination1",
+    "ResourceType": "AWS::SQS::QueuePolicy"
+  }
+]

integration/resources/templates/combination/connector_event_rule_to_lambda_write_multiple.yaml

@@ -0,0 +1,112 @@
+Resources:
+  TriggerFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: nodejs14.x
+      Handler: index.handler
+      Timeout: 10  # in case eb has delay
+      InlineCode: |
+        const AWS = require('aws-sdk');
+
+        exports.handler = async (event) => {
+          const eb = new AWS.EventBridge();
+          const response = await eb.putEvents({
+            Entries: [{
+              Source: process.env.EVENT_SOURCE,
+              Detail: "{}",
+              DetailType: "Test",
+            }]
+          }).promise();
+
+          const sqs = new AWS.SQS();
+          const data = await sqs.receiveMessage({
+            QueueUrl: process.env.QUEUE_URL,
+            WaitTimeSeconds: 5,
+          }).promise();
+
+          if (data.Messages.length == 0) {
+            throw 'No messages in the queue!';
+          }
+        };
+      Environment:
+        Variables:
+          QUEUE_URL: !Ref VerificationQueue
+          EVENT_SOURCE: !Sub '${AWS::StackName}-test-event'
+      Policies:
+      - EventBridgePutEventsPolicy:
+          EventBusName: default
+      - SQSPollerPolicy:
+          QueueName: !GetAtt VerificationQueue.QueueName
+
+  EventRule:
+    Type: AWS::Events::Rule
+    Properties:
+      Description: !Sub 'EventRule-${AWS::StackName}'
+      EventPattern:
+        source:
+        - !Sub '${AWS::StackName}-test-event'
+      Targets:
+      - Arn: !GetAtt Function.Arn
+        Id: Target
+      - Arn: !GetAtt Function2.Arn
+        Id: Target2
+
+  Function:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: nodejs14.x
+      Handler: index.handler
+      InlineCode: |
+        const AWS = require('aws-sdk');
+
+        exports.handler = async (event) => {
+          const sqs = new AWS.SQS();
+          await sqs.sendMessage({
+            QueueUrl: process.env.QUEUE_URL,
+            MessageBody: "test"
+          }).promise();
+        };
+      Environment:
+        Variables:
+          QUEUE_URL: !Ref VerificationQueue
+      Policies:
+      - SQSSendMessagePolicy:
+          QueueName: !GetAtt VerificationQueue.QueueName
+
+  Function2:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: nodejs14.x
+      Handler: index.handler
+      InlineCode: |
+        const AWS = require('aws-sdk');
+
+        exports.handler = async (event) => {
+          const sqs = new AWS.SQS();
+          await sqs.sendMessage({
+            QueueUrl: process.env.QUEUE_URL,
+            MessageBody: "test"
+          }).promise();
+        };
+      Environment:
+        Variables:
+          QUEUE_URL: !Ref VerificationQueue
+      Policies:
+      - SQSSendMessagePolicy:
+          QueueName: !GetAtt VerificationQueue.QueueName
+
+  VerificationQueue:
+    Type: AWS::SQS::Queue
+
+  MyConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id: EventRule
+      Destination:
+      - Id: Function
+      - Id: Function2
+      Permissions:
+      - Write
+Metadata:
+  SamTransformTest: true

integration/resources/templates/combination/connector_function_to_bucket_read_multiple.yaml

@@ -0,0 +1,57 @@
+Resources:
+  LambdaRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Statement:
+        - Effect: Allow
+          Action: sts:AssumeRole
+          Principal:
+            Service: lambda.amazonaws.com
+      ManagedPolicyArns:
+      - !Sub arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
+
+  TriggerFunction:
+    Type: AWS::Lambda::Function
+    Properties:
+      Role: !GetAtt LambdaRole.Arn
+      Runtime: nodejs14.x
+      Handler: index.handler
+      Code:
+        ZipFile: |
+          const AWS = require('aws-sdk');
+          var head_params = {
+            Bucket: process.env.BUCKET,
+          };
+          var head_params_2 = {
+            Bucket: process.env.BUCKET2,
+          }
+          exports.handler = async (event) => {
+            console.log('REQUEST RECEIVED:', JSON.stringify(event));
+            var s3 = new AWS.S3();
+            await s3.headBucket(head_params).promise();
+            await s3.headBucket(head_params_2).promise();
+          };
+      Environment:
+        Variables:
+          BUCKET: !Ref Bucket
+          BUCKET2: !Ref Bucket2
+
+  Bucket:
+    Type: AWS::S3::Bucket
+
+  Bucket2:
+    Type: AWS::S3::Bucket
+
+  Connector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id: TriggerFunction
+      Destination:
+      - Id: Bucket
+      - Id: Bucket2
+      Permissions:
+      - Read
+Metadata:
+  SamTransformTest: true

integration/resources/templates/combination/connector_mix_destination.yaml

@@ -0,0 +1,90 @@
+Resources:
+  TriggerFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: nodejs14.x
+      Handler: index.handler
+      Timeout: 10  # in case eb has delay
+      InlineCode: |
+        const AWS = require('aws-sdk');
+
+        exports.handler = async (event) => {
+          const eb = new AWS.EventBridge();
+          const response = await eb.putEvents({
+            Entries: [{
+              Source: process.env.EVENT_SOURCE,
+              Detail: "{}",
+              DetailType: "Test",
+            }]
+          }).promise();
+          await new Promise(resolve => setTimeout(resolve, 5000));
+          const sqs = new AWS.SQS();
+          const data = await sqs.getQueueAttributes({
+            QueueUrl: process.env.QUEUE_URL,
+            AttributeNames: ['ApproximateNumberOfMessages']
+          }).promise();
+
+          if (data.Attributes.ApproximateNumberOfMessages < 2) {
+            throw 'Not enough messages in the queue!';
+          }
+        };
+      Environment:
+        Variables:
+          QUEUE_URL: !Ref VerificationQueue
+          EVENT_SOURCE: !Sub '${AWS::StackName}-test-event'
+      Policies:
+      - EventBridgePutEventsPolicy:
+          EventBusName: default
+      - SQSPollerPolicy:
+          QueueName: !GetAtt VerificationQueue.QueueName
+
+  EventRule:
+    Type: AWS::Events::Rule
+    Properties:
+      Description: !Sub 'EventRule-${AWS::StackName}'
+      EventPattern:
+        source:
+        - !Sub '${AWS::StackName}-test-event'
+      Targets:
+      - Arn: !GetAtt Function.Arn
+        Id: Target
+      - Arn: !GetAtt VerificationQueue.Arn
+        Id: Target2
+
+  Function:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: nodejs14.x
+      Handler: index.handler
+      InlineCode: |
+        const AWS = require('aws-sdk');
+
+        exports.handler = async (event) => {
+          const sqs = new AWS.SQS();
+          await sqs.sendMessage({
+            QueueUrl: process.env.QUEUE_URL,
+            MessageBody: "test"
+          }).promise();
+        };
+      Environment:
+        Variables:
+          QUEUE_URL: !Ref VerificationQueue
+      Policies:
+      - SQSSendMessagePolicy:
+          QueueName: !GetAtt VerificationQueue.QueueName
+
+  VerificationQueue:
+    Type: AWS::SQS::Queue
+
+  MyConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source:
+        Id: EventRule
+      Destination:
+      - Id: Function
+      - Id: VerificationQueue
+      Permissions:
+      - Write
+Metadata:
+  SamTransformTest: true