EventBridgeのパイプ(Pipes)。1対1の連携ならこっちが便利かも？

## はじめに

今回はイベントバスに似た機能、パイプ(Pipes)についてです。

https://docs.aws.amazon.com/ja_jp/eventbridge/latest/userguide/eb-pipes.html

## イベントバスと、パイプの違い。

似たような機能なので、少し比較表を作ってみました。

| 機能 | パイプ (Pipes) | イベントバス (Event Buses) |
|------|--------------|-------------------------|
| 接続タイプ | 1:1 | 多:多 |
| 用途 | 単一ソースから特定ターゲットへ | 複数ソースから複数ターゲットへ |
| イベント変換 | Lambda,StepFunctions等| 入力トランスフォーマ |
| エラーハンドリング | DLQあり | DLQあり |
| イベントストア | なし | サポート(アーカイブ) |

大は小を兼ねる、でイベントバスを使っておけばとりあえず問題ない感じもしますが、イベント変換部分はパイプの方が優れている感じがします。

## 試してみた。

早速作ってみます。

パイプは以下の４Stepで作っていきます。

![スクリーンショット 2025-04-03 5.44.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/6acce606-2254-421d-a58f-a064c503ca71.png)

### ソース

まずは、イベントの発信元を決定します。
ソースは以下が選択可能でした。

* Kinesis
* SQS
* DynamoDB
* AmazonMQ
* セルフマネージド Apache Kafka
* アマゾンMSK
* Confluentクラウド

![スクリーンショット 2025-04-02 4.15.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/cc0e8f01-4183-42b4-9df1-70f7dfe97bb4.png)

今回は、わかりやすいSQSで検証してみます。

![スクリーンショット 2025-04-02 4.18.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/7da4fa4b-2434-4f5e-bfd6-60c1830f4d47.png)

### フィルタリング

ソースのイベントに対して、以降の処理対象にするかをフィルタリングをすることができます。

事前に選んだソースごとにサンプルイベントを作ってくれていて、動作確認も可能です。

今回は、`body`に`Test`が前方一致で含まれているか、でフィルタリングします。

```
{
  "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
  "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
  "body": "Test message.",
  "attributes": {
    "ApproximateReceiveCount": "1",
    "SentTimestamp": "1545082649183",
    "SenderId": "AIDAIENQZJOLO23YVJ4VO",
    "ApproximateFirstReceiveTimestamp": "1545082649185"
  },
  "messageAttributes": {},
  "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
  "eventSource": "aws:sqs",
  "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
  "awsRegion": "us-east-2"
}
```

![スクリーンショット 2025-04-03 5.57.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/30a73ef9-9c5c-4625-95d8-35d49ae7d1e4.png)

## 強化

実際のターゲットにイベントを送付する前に、イベントデータの加工や、事前処理を実施できたりします。

ここはとても便利だなと思います。

今回はLambdaを設定しました。

![スクリーンショット 2025-04-02 4.34.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/03b6f33b-2842-40c7-984d-2aa0504f6c98.png)

Lambdaは受け取ったイベントに対し、`lambdacheckd`って項目を入れるだけです。

```js
export const handler = async (event) => {
  console.log(event);
  return {...event, lambdacheckd : true};
};
```

### ターゲット

最後にイベントの最終到達先を指定します。

今回はAPI送信先を指定しています。

![スクリーンショット 2025-04-02 4.35.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/c1154c3a-17b2-4100-836c-213983f0c173.png)

`https://webhook.site/`ですが、簡単にWebhookのエンドポイントを作ってくれ、WEB上で通信結果が確認できるとっても便利なサイトです！

https://webhook.site/#!/

### イベントを送ってみます。

フィルタリングルールに一致するように、`Test Message!`とメッセージを送ってみます。
![スクリーンショット 2025-04-03 6.12.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/e9cdbbe8-183e-44cb-8365-51c6c65db1ac.png)

無事ターゲットが呼ばれています！
`lambdacheckd`の項目も入ってますね。

![スクリーンショット 2025-04-03 6.14.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3894042/9d9b3e83-7d1d-48a1-9b4a-0c9c6ef285fe.png)

## 最後に

パイプは1:1と少し限定的な用途にはなりつつ、設定が簡単かつ、「強化」という便利な機能が備わってました！

使えるケースであれば積極的に使っても良いかなと思いました。