import { Construct } from 'constructs' ; import * as kinesis from "aws-cdk-lib/aws-kinesis" ; import * as lambda from "aws-cdk-lib/aws-lambda" ; import * as lambdaEventSources from "aws-cdk-lib/aws-lambda-event-sources" ; import { App } from "./App" ; import { getFunctionRef , SSTConstruct , isCDKConstruct } from "./Construct" ; import { Function as Fn , FunctionProps , FunctionDefinition } from "./Function" ; import { Permissions } from "./util/permission" ; export interface KinesisStreamProps { readonly kinesisStream ? : kinesis . IStream | kinesis . StreamProps ; readonly consumers ? : { [ consumerName ] : FunctionDefinition | KinesisStreamConsumerProps ; } ; readonly defaultFunctionProps ? : FunctionProps ; } export interface KinesisStreamConsumerProps { readonly function : FunctionDefinition ; readonly consumerProps ? : lambdaEventSources . KinesisEventSourceProps ; } export class KinesisStream extends Construct implements SSTConstruct { public readonly kinesisStream : kinesis . IStream ; private functions : { [ consumerName ] : Fn ; } ; private readonly permissionsAttachedForAllConsumers : Permissions [ ] ; private readonly defaultFunctionProps ? : FunctionProps ; constructor ( scope , id , props ? ) { super ( scope , id ) ; const root = scope . node . root as App ; const { kinesisStream , consumers , defaultFunctionProps } = props || { } ; this . functions = { } ; this . permissionsAttachedForAllConsumers = [ ] ; this . defaultFunctionProps = defaultFunctionProps ; if ( isCDKConstruct ( kinesisStream ) ) { this . kinesisStream = kinesisStream as kinesis . IStream ; } else { const kinesisStreamProps = ( kinesisStream || { } ) as kinesis . StreamProps ; this . kinesisStream = new kinesis . Stream ( this , "Stream" , { streamName : root . logicalPrefixedName ( id ) , ... kinesisStreamProps , } ) ; } if ( consumers ) { Object . keys ( consumers ) . forEach ( ( consumerName ) => this . addConsumer ( this , consumerName , consumers [ consumerName ] ) ) ; } } public get streamArn ( ) : string { return this . kinesisStream . streamArn ; } public get streamName ( ) : string { return this . kinesisStream . streamName ; } public addConsumers ( scope , consumers ) { Object . keys ( consumers ) . forEach ( ( consumerName ) => { this . addConsumer ( scope , consumerName , consumers [ consumerName ] ) ; } ) ; } public attachPermissions ( permissions ) { Object . values ( this . functions ) . forEach ( ( fn ) => fn . attachPermissions ( permissions ) ) ; this . permissionsAttachedForAllConsumers . push ( permissions ) ; } public attachPermissionsToConsumer ( consumerName , permissions ) { if ( ! this . functions [ consumerName ] ) { throw new Error ( `The "${ consumerName }" consumer was not found in the "${ this . node . id }" KinesisStream.` ) ; } this . functions [ consumerName ] . attachPermissions ( permissions ) ; } public getFunction ( consumerName ) { return this . functions [ consumerName ] ; } public getConstructMetadata ( ) { return { type : "KinesisStream" as const , data : { streamName : this . kinesisStream . streamName , consumers : Object . entries ( this . functions ) . map ( ( [ name , fn ] ) => ( { name , fn : getFunctionRef ( fn ) , } ) ) , } , } ; } private addConsumer ( scope , consumerName , consumer ) { let consumerFunction , consumerProps ; if ( ( consumer as KinesisStreamConsumerProps ) . function ) { consumer = consumer as KinesisStreamConsumerProps ; consumerFunction = consumer . function ; consumerProps = consumer . consumerProps ; } else { consumerFunction = consumer as FunctionDefinition ; } consumerProps = { startingPosition : lambda . StartingPosition . LATEST , ... ( consumerProps || { } ) , } ; const fn = Fn . fromDefinition ( scope , consumerName , consumerFunction , this . defaultFunctionProps , `The "defaultFunctionProps" cannot be applied if an instance of a Function construct is passed in. Make sure to define all the consumers using FunctionProps, so the KinesisStream construct can apply the "defaultFunctionProps" to them.` ) ; this . functions [ consumerName ] = fn ; const eventSource = new lambdaEventSources . KinesisEventSource ( this . kinesisStream , consumerProps ) ; fn . addEventSource ( eventSource ) ; this . permissionsAttachedForAllConsumers . forEach ( ( permissions ) => { fn . attachPermissions ( permissions ) ; } ) ; return fn ; } }