feat(sns): add support for subscription filter policy (#2778)

Add a new `filterPolicy` prop to all SNS subscriptions (`aws-sns-subscriptions`).

The filter policy can be defined using the new `SubscriptionFilterPolicy` class.

Moved the creation of the `Subscription` back to `aws-sns` to avoid code repetitions.

packages/@aws-cdk/aws-ecs-patterns/test/ec2/integ.scheduled-ecs-task.lit.expected.json

@@ -506,7 +506,7 @@
         "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionServiceRole94543EDA"
       ]
     },
-    "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionTopicSubscriptionDA5F8A10": {
+    "EcsClusterDefaultAutoScalingGroupLifecycleHookDrainHookTopicFunctionA8966A35": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "lambda",

packages/@aws-cdk/aws-ecs/test/ec2/integ.lb-awsvpc-nw.expected.json

@@ -671,7 +671,7 @@
         "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionServiceRole94543EDA"
       ]
     },
-    "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionTopicSubscriptionDA5F8A10": {
+    "EcsClusterDefaultAutoScalingGroupLifecycleHookDrainHookTopicFunctionA8966A35": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "lambda",

packages/@aws-cdk/aws-ecs/test/ec2/integ.lb-bridge-nw.expected.json

@@ -692,7 +692,7 @@
         "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionServiceRole94543EDA"
       ]
     },
-    "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionTopicSubscriptionDA5F8A10": {
+    "EcsClusterDefaultAutoScalingGroupLifecycleHookDrainHookTopicFunctionA8966A35": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "lambda",

packages/@aws-cdk/aws-ecs/test/ec2/integ.sd-awsvpc-nw.expected.json

@@ -671,7 +671,7 @@
         "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionServiceRole94543EDA"
       ]
     },
-    "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionTopicSubscriptionDA5F8A10": {
+    "EcsClusterDefaultAutoScalingGroupLifecycleHookDrainHookTopicFunctionA8966A35": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "lambda",
@@ -946,4 +946,4 @@
       }
     }
   }
-}
+}

packages/@aws-cdk/aws-ecs/test/ec2/integ.sd-bridge-nw.expected.json

@@ -671,7 +671,7 @@
         "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionServiceRole94543EDA"
       ]
     },
-    "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionTopicSubscriptionDA5F8A10": {
+    "EcsClusterDefaultAutoScalingGroupLifecycleHookDrainHookTopicFunctionA8966A35": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "lambda",
@@ -910,4 +910,4 @@
       }
     }
   }
-}
+}

packages/@aws-cdk/aws-events-targets/test/codebuild/integ.project-events.expected.json

@@ -327,7 +327,7 @@
     "MyQueueE6CA6235": {
       "Type": "AWS::SQS::Queue"
     },
-    "MyQueueMyTopicSubscriptionEB66AD1B": {
+    "MyTopicMyQueueFA241964": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "sqs",

packages/@aws-cdk/aws-events-targets/test/ecs/integ.event-ec2-task.lit.expected.json

@@ -506,7 +506,7 @@
         "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionServiceRole94543EDA"
       ]
     },
-    "EcsClusterDefaultAutoScalingGroupDrainECSHookFunctionTopicSubscriptionDA5F8A10": {
+    "EcsClusterDefaultAutoScalingGroupLifecycleHookDrainHookTopicFunctionA8966A35": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "lambda",
@@ -1197,4 +1197,4 @@
       "Description": "Artifact hash for asset \"aws-ecs-integ-ecs/AdoptEcrRepositorydbc60defc59544bcaa5c28c95d68f62c/Code\""
     }
   }
-}
+}

packages/@aws-cdk/aws-events-targets/test/sns/integ.sns-event-rule-target.expected.json

@@ -57,7 +57,7 @@
     "MyQueueE6CA6235": {
       "Type": "AWS::SQS::Queue"
     },
-    "MyQueueMyTopicSubscriptionEB66AD1B": {
+    "MyTopicMyQueueFA241964": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "sqs",
@@ -108,4 +108,4 @@
       }
     }
   }
-}
+}

packages/@aws-cdk/aws-lambda-event-sources/test/integ.sns.expected.json

@@ -60,7 +60,7 @@
         "FServiceRole3AC82EE1"
       ]
     },
-    "FTSubscription775EAF05": {
+    "TF2453034D": {
       "Type": "AWS::SNS::Subscription",
       "Properties": {
         "Protocol": "lambda",

packages/@aws-cdk/aws-sns-subscriptions/lib/email.ts

@@ -1,10 +1,10 @@
 import sns = require('@aws-cdk/aws-sns');
-import { Construct } from '@aws-cdk/cdk';
+import { SubscriptionProps } from './subscription';
 
 /**
  * Options for email subscriptions.
  */
-export interface EmailSubscriptionProps {
+export interface EmailSubscriptionProps extends SubscriptionProps {
   /**
    * Indicates if the full notification JSON should be sent to the email
    * address or just the message text.
@@ -23,11 +23,12 @@ export class EmailSubscription implements sns.ITopicSubscription {
   constructor(private readonly emailAddress: string, private readonly props: EmailSubscriptionProps = {}) {
   }
 
-  public bind(scope: Construct, topic: sns.ITopic): void {
-    new sns.Subscription(scope, this.emailAddress, {
-      topic,
+  public bind(_topic: sns.ITopic): sns.TopicSubscriptionConfig {
+    return {
+      subscriberId: this.emailAddress,
       endpoint: this.emailAddress,
-      protocol: this.props.json ? sns.SubscriptionProtocol.EMAIL_JSON : sns.SubscriptionProtocol.EMAIL
-    });
+      protocol: this.props.json ? sns.SubscriptionProtocol.EMAIL_JSON : sns.SubscriptionProtocol.EMAIL,
+      filterPolicy: this.props.filterPolicy,
+    };
   }
 }

packages/@aws-cdk/aws-sns-subscriptions/lib/index.ts

@@ -1,3 +1,4 @@
+export * from './subscription';
 export * from './email';
 export * from './lambda';
 export * from './sqs';

packages/@aws-cdk/aws-sns-subscriptions/lib/lambda.ts

@@ -2,37 +2,38 @@ import iam = require('@aws-cdk/aws-iam');
 import lambda = require('@aws-cdk/aws-lambda');
 import sns = require('@aws-cdk/aws-sns');
 import { Construct } from '@aws-cdk/cdk';
+import { SubscriptionProps } from './subscription';
 
+/**
+ * Properties for a Lambda subscription
+ */
+export interface LambdaSubscriptionProps extends SubscriptionProps {
+
+}
 /**
  * Use a Lambda function as a subscription target
  */
 export class LambdaSubscription implements sns.ITopicSubscription {
-  constructor(private readonly fn: lambda.IFunction) {
+  constructor(private readonly fn: lambda.IFunction, private readonly props: LambdaSubscriptionProps = {}) {
   }
 
-  public bind(_scope: Construct, topic: sns.ITopic): void {
+  public bind(topic: sns.ITopic): sns.TopicSubscriptionConfig {
     // Create subscription under *consuming* construct to make sure it ends up
     // in the correct stack in cases of cross-stack subscriptions.
     if (!Construct.isConstruct(this.fn)) {
       throw new Error(`The supplied lambda Function object must be an instance of Construct`);
     }
 
-    // we use the target name as the subscription's. there's no meaning to
-    // subscribing the same queue twice on the same topic.
-    const subscriptionName = topic.node.id + 'Subscription';
-    if (this.fn.node.tryFindChild(subscriptionName)) {
-      throw new Error(`A subscription between the topic ${topic.node.id} and the lambda ${this.fn.node.id} already exists`);
-    }
-
-    new sns.Subscription(this.fn, subscriptionName, {
-      topic,
-      endpoint: this.fn.functionArn,
-      protocol: sns.SubscriptionProtocol.LAMBDA,
-    });
-
     this.fn.addPermission(topic.node.id, {
       sourceArn: topic.topicArn,
       principal: new iam.ServicePrincipal('sns.amazonaws.com'),
     });
+
+    return {
+      subscriberId: this.fn.node.id,
+      endpoint: this.fn.functionArn,
+      protocol: sns.SubscriptionProtocol.LAMBDA,
+      filterPolicy: this.props.filterPolicy,
+    };
   }
 }

packages/@aws-cdk/aws-sns-subscriptions/lib/sqs.ts

@@ -2,11 +2,12 @@ import iam = require('@aws-cdk/aws-iam');
 import sns = require('@aws-cdk/aws-sns');
 import sqs = require('@aws-cdk/aws-sqs');
 import { Construct } from '@aws-cdk/cdk';
+import { SubscriptionProps } from './subscription';
 
 /**
  * Properties for an SQS subscription
  */
-export interface SqsSubscriptionProps {
+export interface SqsSubscriptionProps extends SubscriptionProps {
   /**
    * The message to the queue is the same as it was sent to the topic
    *
@@ -24,27 +25,13 @@ export class SqsSubscription implements sns.ITopicSubscription {
   constructor(private readonly queue: sqs.IQueue, private readonly props: SqsSubscriptionProps = {}) {
   }
 
-  public bind(_scope: Construct, topic: sns.ITopic): void {
+  public bind(topic: sns.ITopic): sns.TopicSubscriptionConfig {
     // Create subscription under *consuming* construct to make sure it ends up
     // in the correct stack in cases of cross-stack subscriptions.
     if (!Construct.isConstruct(this.queue)) {
       throw new Error(`The supplied Queue object must be an instance of Construct`);
     }
 
-    // we use the queue name as the subscription's. there's no meaning to
-    // subscribing the same queue twice on the same topic.
-    const subscriptionName = topic.node.id + 'Subscription';
-    if (this.queue.node.tryFindChild(subscriptionName)) {
-      throw new Error(`A subscription between the topic ${topic.node.id} and the queue ${this.queue.node.id} already exists`);
-    }
-
-    new sns.Subscription(this.queue, subscriptionName, {
-      topic,
-      endpoint: this.queue.queueArn,
-      protocol: sns.SubscriptionProtocol.SQS,
-      rawMessageDelivery: this.props.rawMessageDelivery,
-    });
-
     // add a statement to the queue resource policy which allows this topic
     // to send messages to the queue.
     this.queue.addToResourcePolicy(new iam.PolicyStatement({
@@ -55,5 +42,13 @@ export class SqsSubscription implements sns.ITopicSubscription {
         ArnEquals: { 'aws:SourceArn': topic.topicArn }
       }
     }));
+
+    return {
+      subscriberId: this.queue.node.id,
+      endpoint: this.queue.queueArn,
+      protocol: sns.SubscriptionProtocol.SQS,
+      rawMessageDelivery: this.props.rawMessageDelivery,
+      filterPolicy: this.props.filterPolicy,
+    };
   }
 }

packages/@aws-cdk/aws-sns-subscriptions/lib/subscription.ts

@@ -0,0 +1,10 @@
+import sns = require('@aws-cdk/aws-sns');
+
+export interface SubscriptionProps {
+  /**
+   * The filter policy.
+   *
+   * @default - all messages are delivered
+   */
+  readonly filterPolicy?: { [attribute: string]: sns.SubscriptionFilter };
+}

packages/@aws-cdk/aws-sns-subscriptions/lib/url.ts

@@ -1,10 +1,10 @@
 import sns = require('@aws-cdk/aws-sns');
-import { Construct } from '@aws-cdk/cdk';
+import { SubscriptionProps } from './subscription';
 
 /**
  * Options for URL subscriptions.
  */
-export interface UrlSubscriptionProps {
+export interface UrlSubscriptionProps extends SubscriptionProps {
   /**
    * The message to the queue is the same as it was sent to the topic
    *
@@ -29,12 +29,13 @@ export class UrlSubscription implements sns.ITopicSubscription {
     }
   }
 
-  public bind(scope: Construct, topic: sns.ITopic): void {
-    new sns.Subscription(scope, this.url, {
-      topic,
+  public bind(_topic: sns.ITopic): sns.TopicSubscriptionConfig {
+    return {
+      subscriberId: this.url,
       endpoint: this.url,
       protocol: this.url.startsWith('https:') ? sns.SubscriptionProtocol.HTTPS : sns.SubscriptionProtocol.HTTP,
       rawMessageDelivery: this.props.rawMessageDelivery,
-    });
+      filterPolicy: this.props.filterPolicy,
+    };
   }
 }