はじめに

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

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

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

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

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

試してみた。

早速作ってみます。

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

ソース

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

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

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

フィルタリング

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

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

今回は、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"
}

強化

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

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

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

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

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

ターゲット

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

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

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

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

フィルタリングルールに一致するように、Test Message!とメッセージを送ってみます。

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

最後に

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

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