Amazon EventBridge Pipes Sources Construct Library

Version 2.275.0-alpha.0

Separate package

@aws-cdk/aws-pipes-sources-alpha is not part of aws-cdk-lib — install it separately:

npm install @aws-cdk/aws-pipes-sources-alpha
pip install aws-cdk.aws-pipes-sources-alpha

This module is experimental (alpha): its API may change in any release, without a major version bump.

cdk-constructs: Experimental

The APIs of higher level constructs in this module are experimental and under active development. They are subject to non-backward compatible changes or removal in any future version. These are not subject to the Semantic Versioning model and breaking changes will be announced in the release notes. This means that while you may use them, you may need to update your source code when upgrading to a newer version of this package.

EventBridge Pipes Sources let you create a source for a EventBridge Pipe.

For more details see the service documentation:

Documentation

Pipe sources

Pipe sources are the starting point of a EventBridge Pipe. They are the source of the events that are sent to the pipe.

Amazon SQS

A SQS message queue can be used as a source for a pipe. The queue will be polled for new messages and the messages will be sent to the pipe.

declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;

const pipeSource = new sources.SqsSource(sourceQueue);

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: pipeSource,
  target: new SqsTarget(targetQueue)
});

python

# source_queue: sqs.Queue
# target_queue: sqs.Queue


pipe_source = sources.SqsSource(source_queue)

pipe = pipes.Pipe(self, "Pipe",
    source=pipe_source,
    target=SqsTarget(target_queue)
)

java

Queue sourceQueue;
Queue targetQueue;


SqsSource pipeSource = new SqsSource(sourceQueue);

Pipe pipe = Pipe.Builder.create(this, "Pipe")
        .source(pipeSource)
        .target(new SqsTarget(targetQueue))
        .build();

csharp

Queue sourceQueue;
Queue targetQueue;


var pipeSource = new SqsSource(sourceQueue);

var pipe = new Pipe(this, "Pipe", new PipeProps {
    Source = pipeSource,
    Target = new SqsTarget(targetQueue)
});

var sourceQueue Queue
var targetQueue Queue


pipeSource := sources.NewSqsSource(sourceQueue)

pipe := pipes.NewPipe(this, jsii.String("Pipe"), &PipeProps{
	Source: pipeSource,
	Target: awscdkpipestargetsalpha.NewSqsTarget(targetQueue),
})

The polling configuration can be customized:

declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;

const pipeSource = new sources.SqsSource(sourceQueue, {
  batchSize: 10,
  maximumBatchingWindow: cdk.Duration.seconds(10)
});

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: pipeSource,
  target: new SqsTarget(targetQueue)
});

python

# source_queue: sqs.Queue
# target_queue: sqs.Queue


pipe_source = sources.SqsSource(source_queue,
    batch_size=10,
    maximum_batching_window=cdk.Duration.seconds(10)
)

pipe = pipes.Pipe(self, "Pipe",
    source=pipe_source,
    target=SqsTarget(target_queue)
)

java

Queue sourceQueue;
Queue targetQueue;


SqsSource pipeSource = SqsSource.Builder.create(sourceQueue)
        .batchSize(10)
        .maximumBatchingWindow(Duration.seconds(10))
        .build();

Pipe pipe = Pipe.Builder.create(this, "Pipe")
        .source(pipeSource)
        .target(new SqsTarget(targetQueue))
        .build();

csharp

Queue sourceQueue;
Queue targetQueue;


var pipeSource = new SqsSource(sourceQueue, new SqsSourceParameters {
    BatchSize = 10,
    MaximumBatchingWindow = Duration.Seconds(10)
});

var pipe = new Pipe(this, "Pipe", new PipeProps {
    Source = pipeSource,
    Target = new SqsTarget(targetQueue)
});

var sourceQueue Queue
var targetQueue Queue


pipeSource := sources.NewSqsSource(sourceQueue, &SqsSourceParameters{
	BatchSize: jsii.Number(10),
	MaximumBatchingWindow: cdk.Duration_Seconds(jsii.Number(10)),
})

pipe := pipes.NewPipe(this, jsii.String("Pipe"), &PipeProps{
	Source: pipeSource,
	Target: awscdkpipestargetsalpha.NewSqsTarget(targetQueue),
})

Amazon Kinesis

A Kinesis stream can be used as a source for a pipe. The stream will be polled for new messages and the messages will be sent to the pipe.

declare const sourceStream: kinesis.Stream;
declare const targetQueue: sqs.Queue;

const pipeSource = new sources.KinesisSource(sourceStream, {
  startingPosition: sources.KinesisStartingPosition.LATEST,
});

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: pipeSource,
  target: new SqsTarget(targetQueue)
});

python

# source_stream: kinesis.Stream
# target_queue: sqs.Queue


pipe_source = sources.KinesisSource(source_stream,
    starting_position=sources.KinesisStartingPosition.LATEST
)

pipe = pipes.Pipe(self, "Pipe",
    source=pipe_source,
    target=SqsTarget(target_queue)
)

java

Stream sourceStream;
Queue targetQueue;


KinesisSource pipeSource = KinesisSource.Builder.create(sourceStream)
        .startingPosition(KinesisStartingPosition.LATEST)
        .build();

Pipe pipe = Pipe.Builder.create(this, "Pipe")
        .source(pipeSource)
        .target(new SqsTarget(targetQueue))
        .build();

csharp

Stream sourceStream;
Queue targetQueue;


var pipeSource = new KinesisSource(sourceStream, new KinesisSourceParameters {
    StartingPosition = KinesisStartingPosition.LATEST
});

var pipe = new Pipe(this, "Pipe", new PipeProps {
    Source = pipeSource,
    Target = new SqsTarget(targetQueue)
});

var sourceStream Stream
var targetQueue Queue


pipeSource := sources.NewKinesisSource(sourceStream, &KinesisSourceParameters{
	StartingPosition: sources.KinesisStartingPosition_LATEST,
})

pipe := pipes.NewPipe(this, jsii.String("Pipe"), &PipeProps{
	Source: pipeSource,
	Target: awscdkpipestargetsalpha.NewSqsTarget(targetQueue),
})

Amazon DynamoDB

A DynamoDB stream can be used as a source for a pipe. The stream will be polled for new messages and the messages will be sent to the pipe.

const table = new ddb.TableV2(this, 'MyTable', {
  partitionKey: {
    name: 'id',
    type: ddb.AttributeType.STRING,
  },
  dynamoStream: ddb.StreamViewType.NEW_IMAGE,
});
declare const targetQueue: sqs.Queue;

const pipeSource = new sources.DynamoDBSource(table, {
  startingPosition: sources.DynamoDBStartingPosition.LATEST,
});

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: pipeSource,
  target: new SqsTarget(targetQueue)
});

python

# target_queue: sqs.Queue
table = ddb.TableV2(self, "MyTable",
    partition_key=ddb.Attribute(
        name="id",
        type=ddb.AttributeType.STRING
    ),
    dynamo_stream=ddb.StreamViewType.NEW_IMAGE
)

pipe_source = sources.DynamoDBSource(table,
    starting_position=sources.DynamoDBStartingPosition.LATEST
)

pipe = pipes.Pipe(self, "Pipe",
    source=pipe_source,
    target=SqsTarget(target_queue)
)

java

Queue targetQueue;
TableV2 table = TableV2.Builder.create(this, "MyTable")
        .partitionKey(Attribute.builder()
                .name("id")
                .type(AttributeType.STRING)
                .build())
        .dynamoStream(StreamViewType.NEW_IMAGE)
        .build();

DynamoDBSource pipeSource = DynamoDBSource.Builder.create(table)
        .startingPosition(DynamoDBStartingPosition.LATEST)
        .build();

Pipe pipe = Pipe.Builder.create(this, "Pipe")
        .source(pipeSource)
        .target(new SqsTarget(targetQueue))
        .build();

csharp

Queue targetQueue;
var table = new TableV2(this, "MyTable", new TablePropsV2 {
    PartitionKey = new Attribute {
        Name = "id",
        Type = AttributeType.STRING
    },
    DynamoStream = StreamViewType.NEW_IMAGE
});

var pipeSource = new DynamoDBSource(table, new DynamoDBSourceParameters {
    StartingPosition = DynamoDBStartingPosition.LATEST
});

var pipe = new Pipe(this, "Pipe", new PipeProps {
    Source = pipeSource,
    Target = new SqsTarget(targetQueue)
});

var targetQueue Queue
table := ddb.NewTableV2(this, jsii.String("MyTable"), &TablePropsV2{
	PartitionKey: &Attribute{
		Name: jsii.String("id"),
		Type: ddb.AttributeType_STRING,
	},
	DynamoStream: ddb.StreamViewType_NEW_IMAGE,
})

pipeSource := sources.NewDynamoDBSource(table, &DynamoDBSourceParameters{
	StartingPosition: sources.DynamoDBStartingPosition_LATEST,
})

pipe := pipes.NewPipe(this, jsii.String("Pipe"), &PipeProps{
	Source: pipeSource,
	Target: awscdkpipestargetsalpha.NewSqsTarget(targetQueue),
})

Types

Name	Kind	Summary
`DynamoDBSource`	Class	A source that reads from an DynamoDB stream.
`DynamoDBSourceParameters`	Struct	Parameters for the DynamoDB source.
`DynamoDBStartingPosition`	Enum	The position in a DynamoDB stream from which to start reading.
`KinesisSource`	Class	A source that reads from Kinesis.
`KinesisSourceParameters`	Struct	Parameters for the Kinesis source.
`KinesisStartingPosition`	Enum	The position in a Kinesis stream from which to start reading.
`OnPartialBatchItemFailure`	Enum	Define how to handle item process failures.
`SqsSource`	Class	A source that reads from an SQS queue.
`SqsSourceParameters`	Struct	Parameters for the SQS source.
`StreamSource`	Class	Streaming sources.
`StreamSourceParameters`	Struct	Base parameters for streaming sources.

Amazon EventBridge Pipes Sources Construct Library ​

Pipe sources ​

Amazon SQS ​

Amazon Kinesis ​

Amazon DynamoDB ​

Types ​

Amazon EventBridge Pipes Sources Construct Library

Pipe sources

Amazon SQS

Amazon Kinesis

Amazon DynamoDB

Types