# EventBridge Pipes

Source: /aws/services/pipes/

## Introduction

EventBridge Pipes allows users to create point-to-point integrations between event producers and consumers with transform, filter and enrichment steps.
Pipes are particularly useful for scenarios involving real-time data processing, application integration, and automated workflows, while simplifying the process of routing events between AWS services.
Pipes offer a point-to-point connection from one source to one target (one-to-one).
In contrast, EventBridge Event Bus offers a one-to-many integration where an event router delivers one event to zero or more destinations.

LocalStack allows you to use the Pipes APIs in your local environment to create Pipes with SQS queues and Kinesis streams as source and target.
You can also filter events using EventBridge event patterns and enrich events using Lambda.

The supported APIs are available on our [API Coverage section](#api-coverage), which provides information on the extent of Pipe's integration with LocalStack.

:::note
The implementation of EventBridge Pipes is currently in **preview** stage and under active development.
If you would like support for more APIs or report bugs, please make a request on [GitHub Discussion](https://github.com/orgs/localstack/discussions/new/choose).
:::

## Getting started

This guide is designed for users new to EventBridge Pipes and assumes basic knowledge of the AWS CLI and our [`lstk aws`](/aws/developer-tools/running-localstack/lstk/cloud-and-iac-commands/#aws) command.

Start your LocalStack container using your preferred method.
We will demonstrate how to create a Pipe with SQS queues as source and target, and send events to the source queue which will be routed to the target queue.

### Create an SQS queue

Create two SQS queues that will be used as source and target for the Pipe.
Run the following command to create a queue using the [`CreateQueue`](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_CreateQueue.html) API:

```bash
lstk aws sqs create-queue --queue-name source-queue
lstk aws sqs create-queue --queue-name target-queue
```

You can fetch their queue ARNs using the [`GetQueueAttributes`](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_GetQueueAttributes.html) API:

```bash
SOURCE_QUEUE_ARN=$(lstk aws sqs get-queue-attributes --queue-url http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/source-queue --attribute-names QueueArn --output text)
TARGET_QUEUE_ARN=$(lstk aws sqs get-queue-attributes --queue-url http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/target-queue --attribute-names QueueArn --output text)
```

### Create a Pipe

You can now create a Pipe, using the [`CreatePipe`](https://docs.aws.amazon.com/eventbridge/latest/APIReference/API_CreatePipe.html) API.
Run the following command, by specifying the source and target queue ARNs we created earlier:

```bash
lstk aws pipes create-pipe --name sample-pipe \
        --source $SOURCE_QUEUE_ARN \
        --target $TARGET_QUEUE_ARN \
        --role-arn arn:aws:iam::000000000000:role/pipes-role
```

```bash title="Output"
{
    "Arn": "arn:aws:pipes:us-east-1:000000000000:pipe/sample-pipe",
    "CreationTime": "2024-01-26T11:55:27.069088+05:30",
    "CurrentState": "CREATING",
    "DesiredState": "RUNNING",
    "LastModifiedTime": "2024-01-26T11:55:27.069088+05:30",
    "Name": "sample-pipe"
}
```

### Describe the Pipe

You can use the [`DescribePipe`](https://docs.aws.amazon.com/eventbridge/latest/APIReference/API_DescribePipe.html) API to get information about the Pipe:

```bash
lstk aws pipes describe-pipe --name sample-pipe
```

```bash title="Output"
{
    "Arn": "arn:aws:pipes:us-east-1:000000000000:pipe/sample-pipe",
    "CreationTime": "2024-01-26T11:55:27.069088+05:30",
    "CurrentState": "RUNNING",
    "DesiredState": "RUNNING",
    "EnrichmentParameters": {},
    "LastModifiedTime": "2024-01-26T11:55:27.069088+05:30",
    "Name": "sample-pipe",
    "RoleArn": "arn:aws:iam::000000000000:role/pipe-role",
    "Source": "arn:aws:sqs:us-east-1:000000000000:source-queue",
    "SourceParameters": {
        "SqsQueueParameters": {
            "BatchSize": 10
        }
    },
    "StateReason": "USER_INITIATED",
    "Tags": {},
    "Target": "arn:aws:sqs:us-east-1:000000000000:target-queue",
    "TargetParameters": {}
}
```

### Send events to the source queue

You can now send events to the source queue, which will be routed to the target queue.
Run the following command to send an event to the source queue:

```bash
lstk aws sqs send-message \
    --queue-url http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/source-queue \
    --message-body "message-1"
```

### Receive events from the target queue

You can fetch the message from the target queue using the [`ReceiveMessage`](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html) API:

```bash
lstk aws sqs receive-message \
    --queue-url http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/target-queue
```

## Resource Browser

The LocalStack Web Application provides a Resource Browser for managing EventBridge Pipes.
You can access the Resource Browser by opening the LocalStack Web Application in your browser, navigating to the **Resource Browser** section, and then clicking on **EventBridge Pipes** under the **App Integration** section.

![EventBridge Pipes Resource Browser](/images/aws/pipes-resource-browser.png)

The Resource Browser for EventBridge Pipes in LocalStack allows you to perform the following actions:

1. **Create a Pipe**: Click on the **Create Pipe** button to set up a new pipe with a source and target service, filter criteria, and more.
2. **View Pipe Details**: Click on the pipe name to view detailed information, including source, target, batch size, state, and more.
3. **Delete a Pipe**: Select a pipe and click on the **Actions** dropdown menu, followed by **Remove Selected**, to delete the pipe.

## Supported sources

LocalStack supports the following [sources](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-source.html) for Pipes:

* Amazon DynamoDB stream
* Amazon Kinesis stream
* Amazon SQS queue

Please create a feature request on [GitHub](https://github.com/orgs/localstack/discussions) if you miss support for
Amazon MQ broker,
Amazon MSK stream,
or Apache Kafka stream.

## Supported enrichments

LocalStack supports the following [enrichments](https://docs.aws.amazon.com/eventbridge/latest/userguide/pipes-enrichment.html) for Pipes:

* Lambda function

Please create a feature request on [GitHub](https://github.com/orgs/localstack/discussions) if you miss support for
API destination,
Amazon API Gateway,
or Step Functions state machine

## Supported targets

LocalStack supports the following [targets](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-target.html) for Pipes:

* EventBride bus
* Kinesis stream
* Lambda function (SYNC or ASYNC)
* Amazon SNS topic
* Amazon SQS queue
* Step Functions state machine
  * Standard workflows (ASYNC)

Please create a feature request on [GitHub](https://github.com/orgs/localstack/discussions) if you miss support for
API destination,
API Gateway,
Batch job queue,
CloudWatch log group,
ECS task,
Firehose delivery stream,
Inspector assessment template,
Redshift cluster data API queries,
SageMaker Pipeline,
Step Functions state machine: Express workflows (SYNC or ASYNC),
or Timestream for LiveAnalytics table.

## Input transformation

LocalStack supports target and enrichment input transformation for Pipes using JSONPath. Wildcards (*) are also supported.

## Supported log destinations

LocalStack supports the following [log destinations](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-logs.html) for detailed Pipes logging:

* CloudWatch Logs

Please create a feature request on [GitHub](https://github.com/orgs/localstack/discussions) if you miss support for
Firehose stream logs,
or Amazon S3 logs.

## Current Limitations

The EventBridge Pipes implementation in LocalStack is currently in preview stage and has the following limitations:

- Lack of concurrency support (i.e., ParallelizationFactor), resulting in slower processing in high-throughput scenarios.
- Lack of lifecycle management for pipe states (i.e., missing tests for state transitions).
- Lack of re-sharding support when polling from Kinesis and DynamoDB streams.
- Batch handling behavior may have parity issues (e.g., batch flushing rules by size, length, time, etc. are not implemented).

## API Coverage


### EventBridge Pipes API coverage

Source service: `pipes`. 10 of 10 tracked operations are implemented.

Service documentation: /aws/services/pipes/
License availability: available starting with the Ultimate plan. See /aws/licensing/ for current plan details.

| Operation | Status |
| --- | --- |
| CreatePipe | Implemented |
| DeletePipe | Implemented |
| DescribePipe | Implemented |
| ListPipes | Implemented |
| ListTagsForResource | Implemented |
| StartPipe | Implemented |
| StopPipe | Implemented |
| TagResource | Implemented |
| UntagResource | Implemented |
| UpdatePipe | Implemented |