キューコンシューマー向けに注文イベントを変換する

はじめに

プロデューサーは完全な注文イベントを送りますが、コンシューマーに必要なのは注文 ID と数量だけです。ルーティングフィルターを保持しながら、2 つのイベントをコンパクトなキューメッセージに変換します。

EventBridge は受信イベントを照合し、メッセージを保持するキューである SQS にルーティングします。CLI アクセスは設定済みで、個人の AWS アカウントやアクセスキーは不要です。

先に EventBridge で注文イベントをルーティングする を完了してください。独立した VM でこのパイプラインを構築します。以前のバス、ルール、キューは再利用しません。

認定試験との関連

ラボの概要

独立したルーティングリソースを準備する

このステップでは、出荷処理ジョブ用のカスタムバスと空のキューを作成します。

上部の AWS View を使い、CLI クエリをこのラボの実際のリソースと結果と比較します。提供された参照データを保持してください。

プロデューサーのイベントには、ルーティングメタデータ、顧客情報、ネストされた注文が含まれます。キューコンシューマーに必要なのは、注文 ID と数量だけです。入力変換は、イベントからフィールドを選択して移動先の本文を構築し、プロデューサーのエンベロープへのコンシューマーの依存を減らします。

この新しい作業環境には、設定済みの CLI アクセスと無関係な参照データがあります。EV01 のリソースは再利用しません。プロジェクトディレクトリから始めます。シェルの代入は返された識別子を保存します。--query はレスポンスフィールドを選択し、--output text はそのフィールドを次のコマンドで再利用できるようにします。

cd /home/labex/project
BUS_NAME=labex-ev02-bus

以降の操作で使用する RULE_NAME を設定します：

RULE_NAME=labex-ev02-orders

演習用イベントバスを作成します：

aws events create-event-bus --name "$BUS_NAME"

返された値を QUEUE_URL に保存します：

QUEUE_URL=$(aws sqs create-queue \
  --queue-name labex-ev02-jobs \
  --query QueueUrl \
  --output text)
echo "$QUEUE_URL"

返された値を QUEUE_ARN に保存します：

QUEUE_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attribute-names QueueArn \
  --query Attributes.QueueArn \
  --output text)
echo "$QUEUE_ARN"

バス ARN はイベントの移動先を識別します。キュー URL はメッセージ操作に使い、キュー ARN はルールターゲットを識別します。初期状態を確認します。

aws events list-rules --event-bus-name "$BUS_NAME"

キュー $QUEUE_URL の属性を確認します：

aws sqs get-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible

ルールもメッセージもありません。上部の AWS View を確認し、同じカスタムバスと空のキューを確認します。準備の検証を実行してください。

入力トランスフォーマーを使うターゲットを接続する

このステップでは、発注された注文を照合し、ルールを認可して、コンパクトなコンシューマーペイロードを定義します。

ルールのパターンは、プロデューサーとイベントカテゴリを照合します。引用符付きのヒアドキュメントは、シェル展開せずに JSON をそのまま書き込み、file:// はそのファイルを CLI リクエストに読み込みます。

EV01 のルーティングと送信元ルールの権限を再利用し、独立したパイプラインを構築します。今回の新しい設計はコンシューマーとの契約です。ターゲットは一致したイベントを、キューのコンシューマーが必要とするフィールドに変換します。どのイベントがターゲットに届くかは、引き続きフィルターが決めます。

cat > order-pattern.json <<'JSON'
{"source":["labex.orders"],"detail-type":["OrderPlaced"]}
JSON

返された値を RULE_ARN に保存します：

RULE_ARN=$(aws events put-rule \
  --name "$RULE_NAME" \
  --event-bus-name "$BUS_NAME" \
  --event-pattern file://order-pattern.json \
  --state ENABLED \
  --query RuleArn \
  --output text)
echo "$RULE_ARN"

キューには、正確な送信元ルールへの権限付与が必要です。自分のキュー ARN とルール ARN を使って、通常のポリシー JSON を書き込みます。その後、--rawfile が、文字列値を持つ SQS の Policy 属性としてエンコードします。

cat > queue-policy.json <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "events.amazonaws.com"
      },
      "Action": "sqs:SendMessage",
      "Resource": "$QUEUE_ARN",
      "Condition": {
        "ArnEquals": {
          "aws:SourceArn": "$RULE_ARN"
        }
      }
    }
  ]
}
EOF

設定を queue-attributes.json に書き込みます：

jq -n --rawfile policy queue-policy.json '{Policy:$policy}' > queue-attributes.json

キュー $QUEUE_URL に属性を適用します：

aws sqs set-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attributes file://queue-attributes.json

InputPathsMap は、JSONPath で選択したフィールドに名前を付けます。$ はイベントのルートを意味し、$.detail.order.id はネストされた注文 ID を選択します。InputTemplate は、山括弧のプレースホルダーでメッセージを構築します。ID のプレースホルダーは JSON 文字列の引用符内にあります。数量は JSON 数値なので引用符はありません。この例は、単純な ID と正の整数の数量を使います。

cat > targets.json <<EOF
[
  {
    "Id": "order-queue",
    "Arn": "$QUEUE_ARN",
    "InputTransformer": {
      "InputPathsMap": {
        "id": "\$.detail.order.id",
        "quantity": "\$.detail.order.quantity"
      },
      "InputTemplate": "{\"id\":\"<id>\",\"quantity\":<quantity>}"
    }
  }
]
EOF

イベントルールの配信ターゲットを適用します：

aws events put-targets \
  --rule "$RULE_NAME" \
  --event-bus-name "$BUS_NAME" \
  --targets file://targets.json

イベントルールの配信ターゲットを確認します：

aws events list-targets-by-rule --rule "$RULE_NAME" --event-bus-name "$BUS_NAME"

FailedEntryCount は 0 です。一覧のターゲットには、自分のキュー ARN と、2 つのマッピングパスおよびテンプレートが含まれます。設定が成功しても、実際の配信テストが必要です。AWS View には有効なルールとトランスフォーマーが表示され、キューは空のままです。接続の検証を実行してください。

2 つの実際のコンシューマーペイロードを比較する

このステップでは、異なる 2 つの発注を発行し、生成されたキューメッセージを確認します。

イベント詳細を通常の JSON オブジェクトとして書き込みます。PutEvents は、各 Detail に JSON エンコードされた文字列を要求します。以下の短い jq コマンドは、そのフィールドだけを変換します。各イベントには、出荷処理のコンシューマーが必要としない合成顧客情報も含まれます。3 つ目のキャンセルイベントは、ルーティングフィルターが引き続き適用されることをテストします。

cat > event-inputs.json <<EOF
[
  {
    "EventBusName": "$BUS_NAME",
    "Source": "labex.orders",
    "DetailType": "OrderPlaced",
    "Detail": {
      "order": {
        "id": "transform-order-a",
        "quantity": 2
      },
      "customer": {
        "email": "synthetic-a@example.test"
      }
    }
  },
  {
    "EventBusName": "$BUS_NAME",
    "Source": "labex.orders",
    "DetailType": "OrderPlaced",
    "Detail": {
      "order": {
        "id": "transform-order-b",
        "quantity": 4
      },
      "customer": {
        "email": "synthetic-b@example.test"
      }
    }
  },
  {
    "EventBusName": "$BUS_NAME",
    "Source": "labex.orders",
    "DetailType": "OrderCancelled",
    "Detail": {
      "order": {
        "id": "cancelled-order",
        "quantity": 9
      }
    }
  }
]
EOF

設定を events.json に書き込みます：

jq 'map(.Detail |= tojson)' event-inputs.json > events.json

提示されたイベントバッチを発行します：

aws events put-events --entries file://events.json

キュー $QUEUE_URL の属性を確認します：

aws sqs get-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible

3 つのイベントはすべて受け付けられますが、一致するのは 2 つの発注だけです。キューには利用可能なメッセージが 2 件あります。可視性 0 で受信すると、確認直後にメッセージを利用可能な状態に保ちます。クエリはメッセージ ID と本文だけを表示し、レシートハンドルを非公開に保ちます。

fromjson は、SQS メッセージ ID を保持しながら各 JSON 本文を読みやすくします。変更されるのは表示出力だけです。

aws sqs receive-message \
  --queue-url "$QUEUE_URL" \
  --max-number-of-messages 10 \
  --visibility-timeout 0 \
  --output json | jq '[.Messages[] | {MessageId, Body: (.Body | fromjson)}]'

本文は {"id":"transform-order-a","quantity":2} と {"id":"transform-order-b","quantity":4} です。順序は評価しません。異なるメッセージ ID を持ち、値は対応するイベントから得られます。どちらにも顧客メール、ルーティングメタデータ、ネストされた order オブジェクトは含まれません。キャンセルされた注文は存在しません。これで証明するのは変換と配信であり、出荷処理の完了やメッセージの処理済み確認ではありません。

AWS View は、実際のコンパクトなキュー本文の隣に、ターゲットのマッピングを表示します。

ペイロードの検証を実行してください。

使い捨てパイプラインを削除する

このステップでは、自分のルーティングリソースを削除し、無関係な状態を保持します。

ルールの前にターゲットを削除し、その後、カスタムバスとキューを削除します。キューには合成された確認用メッセージだけが含まれます。キューの削除は、それらを破棄するものであり、業務処理を行ったとは主張しません。

aws events remove-targets \
  --rule "$RULE_NAME" \
  --event-bus-name "$BUS_NAME" \
  --ids order-queue

演習用イベントルールのみを削除します：

aws events delete-rule --name "$RULE_NAME" --event-bus-name "$BUS_NAME"

演習用イベントバスのみを削除します：

aws events delete-event-bus --name "$BUS_NAME"

演習用キュー $QUEUE_URL を削除します：

aws sqs delete-queue --queue-url "$QUEUE_URL"

成功したネイティブな一覧クエリが、削除の証拠になります。

aws events list-event-buses

残っているイベントルールを確認します：

aws events list-rules --event-bus-name default

残っている SQS キューを確認します：

aws sqs list-queues

演習と無関係な参照レコードを確認します：

aws dynamodb scan --table-name labex-ev02-reference --query Items

ルールのないデフォルトバスだけが残り、キュー URL は存在しません。参照アイテムには引き続き keep unchanged があります。ネットワークエラーや認証エラーは削除の証拠になりません。AWS View は、空のカスタムリソースと保持された参照リソースを表示します。

このラボで作成した通常のファイルを削除します。

rm -f event-inputs.json order-pattern.json queue-policy.json queue-attributes.json targets.json events.json

VM を終了する前に、後片付けの検証を実行してください。

まとめ

ネストされた注文フィールドを、コンパクトな SQS コンシューマーの入力仕様にマッピングし、2 つの実際のイベントから異なる値を確認して、イベントフィルタリングと送信元ルールの権限を保持しました。参照データを保持しながら、使い捨てパイプラインを削除しました。

次のユニットから独立したワークフロー分岐が始まります。その Lambda–DynamoDB の前提知識が必要で、このイベントルーティングのパイプラインには依存しません。

詳しくは、AWS 公式ドキュメントを参照してください：入力パスとターゲットペイロードテンプレート。