feat(lambda-event-sources): expose enabled

Specifically, it enables this flag in the props of which resources use
an underlying Cfn EventSourceMapping, which carries the Enabled flag
itself.
These are currently: SQS, DynamoDb Streams and Kinesis.

Closes #5750

packages/@aws-cdk/aws-lambda-event-sources/README.md

@@ -51,6 +51,7 @@ behavior:
 * __receiveMessageWaitTime__: Will determine [long
   poll](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html)
   duration. The default value is 20 seconds.
+* __enabled__: If If the SQS event source mapping should be enabled. The default is true.
 
 ```ts
 import * as sqs from '@aws-cdk/aws-sqs';
@@ -145,6 +146,7 @@ and add it to your Lambda function. The following parameters will impact Amazon
 * __parallelizationFactor__: The number of batches to concurrently process on each shard.
 * __retryAttempts__: The maximum number of times a record should be retried in the event of failure.
 * __startingPosition__: Will determine where to being consumption, either at the most recent ('LATEST') record or the oldest record ('TRIM_HORIZON'). 'TRIM_HORIZON' will ensure you process all available data, while 'LATEST' will ignore all records that arrived prior to attaching the event source.
+* __enabled__: If If the DynamoDB Streams event source mapping should be enabled. The default is true.
 
 ```ts
 import * as dynamodb from '@aws-cdk/aws-dynamodb';
@@ -188,6 +190,7 @@ behavior:
 * __parallelizationFactor__: The number of batches to concurrently process on each shard.
 * __retryAttempts__: The maximum number of times a record should be retried in the event of failure.
 * __startingPosition__: Will determine where to being consumption, either at the most recent ('LATEST') record or the oldest record ('TRIM_HORIZON'). 'TRIM_HORIZON' will ensure you process all available data, while 'LATEST' will ignore all records that arrived prior to attaching the event source.
+* __enabled__: If If the DynamoDB Streams event source mapping should be enabled. The default is true.
 
 ```ts
 import * as lambda from '@aws-cdk/aws-lambda';

packages/@aws-cdk/aws-lambda-event-sources/lib/sqs.ts

@@ -12,6 +12,13 @@ export interface SqsEventSourceProps {
    * @default 10
    */
   readonly batchSize?: number;
+
+  /**
+   * If the SQS event source mapping should be enabled.
+   *
+   * @default true
+   */
+  readonly enabled?: boolean;
 }
 
 /**
@@ -29,6 +36,7 @@ export class SqsEventSource implements lambda.IEventSource {
   public bind(target: lambda.IFunction) {
     const eventSourceMapping = target.addEventSourceMapping(`SqsEventSource:${this.queue.node.uniqueId}`, {
       batchSize: this.props.batchSize,
+      enabled: this.props.enabled,
       eventSourceArn: this.queue.queueArn,
     });
     this._eventSourceMappingId = eventSourceMapping.eventSourceMappingId;

packages/@aws-cdk/aws-lambda-event-sources/lib/stream.ts

@@ -77,6 +77,13 @@ export interface StreamEventSourceProps {
    * @default Duration.seconds(0)
    */
   readonly maxBatchingWindow?: Duration;
+
+  /**
+   * If the stream event source mapping should be enabled.
+   *
+   * @default true
+   */
+  readonly enabled?: boolean;
 }
 
 /**
@@ -99,6 +106,7 @@ export abstract class StreamEventSource implements lambda.IEventSource {
       retryAttempts: this.props.retryAttempts,
       parallelizationFactor: this.props.parallelizationFactor,
       onFailure: this.props.onFailure,
+      enabled: this.props.enabled,
     };
   }
 }

packages/@aws-cdk/aws-lambda-event-sources/test/test.dynamo.ts

@@ -596,4 +596,30 @@ export = {
 
     test.done();
   },
+
+  'event source disabled'(test: Test) {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const table = new dynamodb.Table(stack, 'T', {
+      partitionKey: {
+        name: 'id',
+        type: dynamodb.AttributeType.STRING,
+      },
+      stream: dynamodb.StreamViewType.NEW_IMAGE,
+    });
+
+    // WHEN
+    fn.addEventSource(new sources.DynamoEventSource(table, {
+      startingPosition: lambda.StartingPosition.LATEST,
+      enabled: false,
+    }));
+
+    //THEN
+    expect(stack).to(haveResource('AWS::Lambda::EventSourceMapping', {
+      'Enabled': false,
+    }));
+
+    test.done();
+  },
 };

packages/@aws-cdk/aws-lambda-event-sources/test/test.kinesis.ts

@@ -210,4 +210,24 @@ export = {
     test.throws(() => eventSource.eventSourceMappingId, /KinesisEventSource is not yet bound to an event source mapping/);
     test.done();
   },
+
+  'event source disabled'(test: Test) {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const stream = new kinesis.Stream(stack, 'S');
+    const eventSource = new sources.KinesisEventSource(stream, {
+      startingPosition: lambda.StartingPosition.LATEST,
+      enabled: false,
+    });
+
+    // WHEN
+    fn.addEventSource(eventSource);
+
+    // THEN
+    expect(stack).to(haveResource('AWS::Lambda::EventSourceMapping', {
+      'Enabled': false,
+    }));
+    test.done();
+  },
 };

packages/@aws-cdk/aws-lambda-event-sources/test/test.sqs.ts

@@ -138,4 +138,23 @@ export = {
     test.throws(() => eventSource.eventSourceMappingId, /SqsEventSource is not yet bound to an event source mapping/);
     test.done();
   },
+
+  'event source disabled'(test: Test) {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const q = new sqs.Queue(stack, 'Q');
+
+    // WHEN
+    fn.addEventSource(new sources.SqsEventSource(q, {
+      enabled: false,
+    }));
+
+    // THEN
+    expect(stack).to(haveResource('AWS::Lambda::EventSourceMapping', {
+      'Enabled': false,
+    }));
+
+    test.done();
+  },
 };