feat(pipes-targets): add EventBridge

packages/@aws-cdk/aws-pipes-targets-alpha/README.md

@@ -29,11 +29,12 @@ Pipe targets are the end point of a EventBridge Pipe.
 
 The following targets are supported:
 
-1. `targets.SqsTarget`: [Send event source to a Queue](#amazon-sqs)
+1. `targets.SqsTarget`: [Send event source to an SQS queue](#amazon-sqs)
 2. `targets.SfnStateMachine`: [Invoke a State Machine from an event source](#aws-step-functions-state-machine)
-3. `targets.LambdaFunction`: [Send event source to a Lambda Function](#aws-lambda-function)
-4. `targets.ApiDestinationTarget`: [Send event source to an EventBridge API Destination](#amazon-eventbridge-api-destination)
+3. `targets.LambdaFunction`: [Send event source to a Lambda function](#aws-lambda-function)
+4. `targets.ApiDestinationTarget`: [Send event source to an EventBridge API destination](#amazon-eventbridge-api-destination)
 5. `targets.KinesisTarget`: [Send event source to a Kinesis data stream](#amazon-kinesis-data-stream)
+6. `targets.EventBridgeTarget`: [Send event soruce to an EventBridge event bus](#amazon-eventbridge-event-bus)
 
 ### Amazon SQS
 
@@ -100,7 +101,6 @@ const pipeTarget = new targets.SfnStateMachine(targetStateMachine,
     }
 );
 
-
 const pipe = new pipes.Pipe(this, 'Pipe', {
     source: new SomeSource(sourceQueue),
     target: pipeTarget
@@ -241,3 +241,35 @@ const pipe = new pipes.Pipe(this, 'Pipe', {
     target: streamTarget,
 });
 ```
+
+### Amazon EventBridge Event Bus
+
+An event bus can be used as a target for a pipe. The event bus will receive the (enriched/filtered) source payload.
+
+```ts
+declare const sourceQueue: sqs.Queue;
+declare const targetEventBus: events.EventBus;
+
+const eventBusTarget = new targets.EventBridgeTarget(targetEventBus);
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+    source: new SqsSource(sourceQueue),
+    target: eventBusTarget,
+});
+```
+
+The input to the target event bus can be transformed:
+
+```ts
+declare const sourceQueue: sqs.Queue;
+declare const targetEventBus: events.EventBus;
+
+const eventBusTarget = new targets.EventBridgeTarget(targetEventBus, {
+  inputTransformation: pipes.InputTransformation.fromObject({ body: "👀" }),
+});
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+    source: new SqsSource(sourceQueue),
+    target: eventBusTarget,
+});
+```

packages/@aws-cdk/aws-pipes-targets-alpha/lib/event-bridge.ts

@@ -0,0 +1,132 @@
+import { IInputTransformation, IPipe, ITarget, TargetConfig } from '@aws-cdk/aws-pipes-alpha';
+import { Token } from 'aws-cdk-lib';
+import { IEventBus } from 'aws-cdk-lib/aws-events';
+import { IRole } from 'aws-cdk-lib/aws-iam';
+
+/**
+ * EventBridge target properties.
+ */
+export interface EventBridgeTargetParameters {
+  /**
+   * The input transformation to apply to the message before sending it to the target.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargetparameters.html#cfn-pipes-pipe-pipetargetparameters-inputtemplate
+   * @default - none
+   */
+  readonly inputTransformation?: IInputTransformation;
+
+  /**
+   * A free-form string used to decide what fields to expect in the event detail.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargeteventbridgeeventbusparameters.html#cfn-pipes-pipe-pipetargeteventbridgeeventbusparameters-detailtype
+   * @default - none
+   */
+  readonly detailType?: string;
+
+  /**
+   * The URL subdomain of the endpoint.
+   *
+   * @example
+   * // if the URL for the endpoint is https://abcde.veo.endpoints.event.amazonaws.com
+   * 'abcde.veo'
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargeteventbridgeeventbusparameters.html#cfn-pipes-pipe-pipetargeteventbridgeeventbusparameters-endpointid
+   * @default - none
+   */
+  readonly endpointId?: string;
+
+  /**
+   * AWS resources, identified by Amazon Resource Name (ARN), which the event primarily concerns.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargeteventbridgeeventbusparameters.html#cfn-pipes-pipe-pipetargeteventbridgeeventbusparameters-resources
+   * @default - none
+   */
+  readonly resources?: string[];
+
+  /**
+   * The source of the event.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargeteventbridgeeventbusparameters.html#cfn-pipes-pipe-pipetargeteventbridgeeventbusparameters-source
+   * @default - none
+   */
+  readonly source?: string;
+
+  /**
+   * The time stamp of the event, per RFC3339.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargeteventbridgeeventbusparameters.html#cfn-pipes-pipe-pipetargeteventbridgeeventbusparameters-time
+   * @default - the time stamp of the PutEvents call
+   */
+  readonly time?: string;
+}
+
+/**
+ * A EventBridge Pipes target that sends messages to an EventBridge event bus.
+ */
+export class EventBridgeTarget implements ITarget {
+  private eventBus: IEventBus;
+  private eventBridgeParameters?: EventBridgeTargetParameters;
+  public readonly targetArn: string;
+
+  constructor(eventBus: IEventBus, parameters?: EventBridgeTargetParameters) {
+    this.eventBus = eventBus;
+    this.targetArn = eventBus.eventBusArn;
+    if (parameters) {
+      this.eventBridgeParameters = parameters;
+      for (const validate of [validateDetailType, validateEndpointId, validateSource, validateTime]) {
+        validate(parameters);
+      }
+    }
+  }
+
+  grantPush(grantee: IRole): void {
+    this.eventBus.grantPutEventsTo(grantee);
+  }
+
+  bind(pipe: IPipe): TargetConfig {
+    if (!this.eventBridgeParameters) {
+      return {
+        targetParameters: {},
+      };
+    }
+
+    return {
+      targetParameters: {
+        inputTemplate: this.eventBridgeParameters.inputTransformation?.bind(pipe).inputTemplate,
+        eventBridgeEventBusParameters: this.eventBridgeParameters,
+      },
+    };
+  }
+}
+
+function validateDetailType({ detailType }: EventBridgeTargetParameters) {
+  if (detailType !== undefined && !Token.isUnresolved(detailType)) {
+    if (detailType.length < 1 || detailType.length > 128) {
+      throw new Error(`Detail type must be between 1 and 128 characters, received ${detailType.length}`);
+    }
+  }
+}
+
+function validateEndpointId({ endpointId }: EventBridgeTargetParameters) {
+  if (endpointId !== undefined && !Token.isUnresolved(endpointId)) {
+    if (endpointId.length < 1 || endpointId.length > 50) {
+      throw new Error(`Endpoint id must be between 1 and 50 characters, received ${endpointId.length}`);
+    }
+  }
+}
+
+function validateSource({ source }: EventBridgeTargetParameters) {
+  if (source !== undefined && !Token.isUnresolved(source)) {
+    if (source.length < 1 || source.length > 256) {
+      throw new Error(`Source must be between 1 and 256 characters, received ${source.length}`);
+    }
+  }
+}
+
+function validateTime({ time }: EventBridgeTargetParameters) {
+  if (time !== undefined && !Token.isUnresolved(time)) {
+    if (time.length < 1 || time.length > 256) {
+      throw new Error(`Time must be between 1 and 256 characters, received ${time.length}`);
+    }
+  }
+}

packages/@aws-cdk/aws-pipes-targets-alpha/lib/index.ts

@@ -1,4 +1,5 @@
 export * from './api-destination';
+export * from './event-bridge';
 export * from './kinesis';
 export * from './lambda';
 export * from './sqs';

packages/@aws-cdk/aws-pipes-targets-alpha/package.json

@@ -89,7 +89,8 @@
     "aws-cdk-lib": "0.0.0",
     "constructs": "^10.0.0",
     "@aws-cdk/aws-pipes-alpha": "0.0.0",
-    "@aws-cdk/integ-tests-alpha": "0.0.0"
+    "@aws-cdk/integ-tests-alpha": "0.0.0",
+    "@aws-cdk/aws-pipes-sources-alpha": "0.0.0"
   },
   "dependencies": {},
   "peerDependencies": {

packages/@aws-cdk/aws-pipes-targets-alpha/test/__snapshots__/event-bridge.test.ts.snap

@@ -0,0 +1,54 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`EventBridge should grant pipe role push access 1`] = `
+{
+  "MyPipeRoleCBC8E9AB": {
+    "Properties": {
+      "AssumeRolePolicyDocument": {
+        "Statement": [
+          {
+            "Action": "sts:AssumeRole",
+            "Effect": "Allow",
+            "Principal": {
+              "Service": "pipes.amazonaws.com",
+            },
+          },
+        ],
+        "Version": "2012-10-17",
+      },
+    },
+    "Type": "AWS::IAM::Role",
+  },
+}
+`;
+
+exports[`EventBridge should grant pipe role push access 2`] = `
+{
+  "MyPipeRoleDefaultPolicy31387C20": {
+    "Properties": {
+      "PolicyDocument": {
+        "Statement": [
+          {
+            "Action": "events:PutEvents",
+            "Effect": "Allow",
+            "Resource": {
+              "Fn::GetAtt": [
+                "MyEventBus251E60F8",
+                "Arn",
+              ],
+            },
+          },
+        ],
+        "Version": "2012-10-17",
+      },
+      "PolicyName": "MyPipeRoleDefaultPolicy31387C20",
+      "Roles": [
+        {
+          "Ref": "MyPipeRoleCBC8E9AB",
+        },
+      ],
+    },
+    "Type": "AWS::IAM::Policy",
+  },
+}
+`;