Промышленная эксплуатация

Это руководство запускает сервис на SQS в AWS: откуда берутся его учётные данные и регион, с какими тайм-аутами он работает и какие права IAM ему нужны.

Учётные данные и регион из окружения

SqsBroker::new() ничего не берёт из кода. При старте сервиса регион и учётные данные находит цепочка AWS SDK по умолчанию: переменные окружения, общие файлы ~/.aws, SSO и роль контейнера или инстанса. Сервису, который работает под ролью IAM в ECS, EKS, Lambda или EC2, конфигурация вообще не нужна.

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

Собственная конфигурация

Макросы

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, the
/// credentials of the `orders-service` profile, read when the first request is signed, and a
/// request timeout above the 20 seconds a long poll waits.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .timeout_config(
            TimeoutConfig::builder()
                .operation_attempt_timeout(Duration::from_secs(25))
                .build(),
        )
        .build()
}

Вручную

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, the
/// credentials of the `orders-service` profile, read when the first request is signed, and a
/// request timeout above the 20 seconds a long poll waits.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .timeout_config(
            TimeoutConfig::builder()
                .operation_attempt_timeout(Duration::from_secs(25))
                .build(),
        )
        .build()
}

SqsBroker::from_config принимает SdkConfig, который собирает сервис. aws_config собирает его из окружения асинхронным загрузчиком, а функция приложения синхронна. Поэтому пример собирает конфигурацию через SdkConfig::builder(), который не выполняет ввода-вывода. Поставщик профилей читает профиль orders-service при первом запросе; профиль с role_arn принимает эту роль. Закреплённая версия поведения не даёт умолчаниям SDK измениться при обновлении SDK. Конфигурация задаёт и тайм-аут запроса, описанный ниже. Брокер сохраняет все её настройки, кроме конечной точки и региона, заданных на самом брокере.

Примеру нужны ещё две зависимости:

Cargo.toml

[dependencies]
aws-config = "1"
aws-sdk-sqs = "1"

Сервис

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    amount_cents: u64,
}

// The queue belongs to the infrastructure, so the descriptor does not create it. It names its
// visibility, so the subscription does not read the queue's own.
#[subscriber(SqsQueue::new("orders").visibility(Duration::from_secs(120)))]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        SqsBroker::from_config(aws_config()),
        |b| {
            b.include(take_order);
        },
    )
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

impl MessageInfo for Order {
    const NAME: &'static str = "Order";
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        SqsBroker::from_config(aws_config()),
        |b| {
            // The queue belongs to the infrastructure, so the descriptor does not create it. It
            // names its visibility, so the subscription does not read the queue's own.
            b.include(
                subscriber("orders", TakeOrder)
                    .map_source(|_| SqsQueue::new("orders").visibility(Duration::from_secs(120)))
                    .build(),
            );
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

Очередь orders относится к инфраструктуре: её создают Terraform, CloudFormation или консоль, вместе с политикой перенаправления и шифрованием. Дескриптор опускает create_if_missing, поэтому отсутствующая очередь останавливает сервис при старте, а не создаётся пустой. Он называет свою видимость, поэтому подписка не читает собственный тайм-аут очереди.

endpoint и region направляют брокер на другую конечную точку, конечную точку VPC или локальный стенд. У брокера, собранного через from_config, они заменяют endpoint_url и region самой конфигурации. Адреса очередей, которые возвращает SQS, переносятся на эту конечную точку.

Тайм-ауты и транспорт

SqsBroker::new() даёт каждому запросу к SQS 25 секунд на попытку, больше 20 секунд, которые может ждать длинный опрос. Собственная конфигурация задаёт это через timeout_config, как в примере; тайм-аут попытки меньше 20 секунд обрывал бы каждый длинный опрос. Неудавшийся запрос SDK повторяет по собственной политике повторов.

Запросы идут через HTTPS-клиент SDK по умолчанию. Обработчик на выделенном потоке (threads(n)) отправляет запросы через собственные клиенты SDK, собранные из той же конфигурации и разделяющие её учётные данные.

Права IAM

Сервису выше нужны такие права на его очередь:

policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "sqs:GetQueueUrl",
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:ChangeMessageVisibility"
      ],
      "Resource": "arn:aws:sqs:eu-west-1:123456789012:orders"
    }
  ]
}

Каждая возможность добавляет свои:

Возможность	Права
подписка, которая не называет видимость	sqs:GetQueueAttributes на очереди
max_attempts и dead_letter	sqs:SetQueueAttributes на очереди, sqs:GetQueueAttributes на DLQ
create_if_missing	sqs:CreateQueue
публикация в очередь	sqs:GetQueueUrl и sqs:SendMessage на очереди
публикация в тему	sns:CreateTopic и sns:Publish на теме
subscribe_queue_to_topic	sns:CreateTopic и sns:Subscribe на теме, sqs:GetQueueUrl, sqs:GetQueueAttributes и sqs:SetQueueAttributes на очереди

Что показывает документ AsyncAPI

С фичей asyncapi документ AsyncAPI сервиса называет один сервер: хост конечной точки брокера, а без неё региональный хост его региона, например sqs.eu-west-1.amazonaws.com. Документ никогда не содержит учётных данных или ARN очереди, поэтому его можно публиковать.

Тесты в файле примера запускают то же приложение внутри процесса; SQS внутри процесса берёт регион из конфигурации и ничего не подписывает. Обвязку разбирает руководство Тестирование сервиса на SQS.