Преобразование событий заказов для потребителя очереди

Введение

Производитель отправляет полное событие заказа, но потребителю нужны только идентификатор заказа и количество. Вы преобразуете два события в компактные сообщения очереди, сохранив фильтр маршрутизации.

EventBridge отбирает входящие события и направляет их в SQS — очередь, которая хранит сообщения. Доступ через CLI уже настроен; личная учётная запись AWS и ключи доступа не нужны.

Сначала выполните Маршрутизацию событий заказов с EventBridge. Постройте этот конвейер в его независимой VM; прежние шина, правило или очередь не используются.

Связь с сертификацией

Обзор лабораторной работы

Подготовка независимых ресурсов маршрутизации

На этом шаге создайте собственную шину и пустую очередь для заданий исполнения заказов.

Используйте AWS View вверху, чтобы сравнивать запросы CLI с реальными ресурсами и результатами этой работы. Сохраните предоставленные эталонные данные.

Событие производителя содержит метаданные маршрутизации, сведения о клиенте и вложенный заказ. Потребителю очереди нужны только идентификатор заказа и количество. Преобразование входных данных выбирает поля события и строит тело назначения, уменьшая зависимость потребителя от конверта производителя.

Эта новая рабочая среда содержит настроенный доступ 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 вверху, чтобы проверить те же собственную шину и пустую очередь. Выполните проверку подготовки.

Подключение цели с преобразователем входных данных

На этом шаге сопоставьте размещённые заказы, авторизуйте правило и определите компактную полезную нагрузку потребителя.

Шаблон правила сопоставляет производителя и категорию события. Here-document в кавычках записывает буквальный 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"

Очереди нужно точное разрешение исходного правила. Запишите обычную политику JSON с ARN своей очереди и правила. Затем --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 выбирает вложенный идентификатор заказа. InputTemplate использует заполнители в угловых скобках для построения сообщения. Заполнитель идентификатора находится внутри кавычек строки JSON; количество — число JSON без кавычек. Пример использует простые идентификаторы и положительные целые количества.

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 равен нулю. Перечисленная цель включает ARN вашей очереди и два пути сопоставления с шаблоном. Успех конфигурации всё ещё требует реальной проверки доставки. AWS View показывает включённое правило и преобразователь; очередь остаётся пустой. Выполните проверку подключения.

Сравнение двух реальных полезных нагрузок потребителя

На этом шаге опубликуйте два разных размещённых заказа и проверьте полученные сообщения очереди.

Запишите подробности событий как обычные объекты JSON. PutEvents требует, чтобы каждый Detail был строкой, закодированной в JSON; короткая команда jq ниже преобразует только это поле. Каждое событие также содержит синтетические сведения о клиенте, которые потребителю исполнения заказов не нужны. Третье событие отмены проверяет сохранение фильтра маршрутизации.

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

Все три события приняты, но совпадают только два размещённых заказа. В очереди два доступных сообщения. Получение с нулевой видимостью оставляет сообщения доступными сразу после проверки. Запрос показывает только идентификаторы сообщений и тела, сохраняя дескрипторы получения приватными.

fromjson делает каждое тело JSON читаемым, сохраняя его идентификатор сообщения SQS. Он меняет только отображаемый вывод.

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}; их порядок не оценивается. У них разные идентификаторы сообщений и значения из соответствующих событий. Ни одно не содержит электронной почты клиента, метаданных маршрутизации или вложенного объекта 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, проверили разные значения двух реальных событий и сохранили фильтрацию событий и разрешения исходного правила. Вы удалили временный конвейер, сохранив эталонные данные.

Следующая работа начинает независимую ветку процессов после её предпосылок Lambda–DynamoDB; она не зависит от этой цепочки событий.

Подробнее см. в документации AWS: Пути входных данных и шаблоны целевой нагрузки.