Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Доставка всем подписчикам разом через SNS

Это руководство публикует сообщение в тему SNS один раз, а SNS доставляет копию в каждую очередь SQS, подписанную на эту тему.

Связка SNS и SQS - это способ, которым сервисы AWS доставляют сообщения всем подписчикам разом: одна тема и по очереди на каждого потребителя. Крейт публикует в темы, а каждую очередь за темой сервис читает обычным образом. Для публикации в SNS нужна фича sns:

Cargo.toml
[dependencies]
ruststream-sqs-sns = { version = "0.7", features = ["sns"] }

Пример - магазин. Он принимает заказ и объявляет о нём в теме, а оплата и доставка читают это объявление каждая из своей очереди.

Ответ в тему​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shop-orders")]
struct PlaceOrder {
    id: u64,
    amount_cents: u64,
}

/// The announcement names the topic, so the handler answers with a bare `reply` clause.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shop-order-events")]
struct OrderPlaced {
    id: u64,
    amount_cents: u64,
}

#[subscriber(SqsQueue::new("shop-orders").create_if_missing(), reply)]
async fn accept(order: &PlaceOrder) -> OrderPlaced {
    tracing::info!(order.id, "order accepted");
    OrderPlaced {
        id: order.id,
        amount_cents: order.amount_cents,
    }
}

accept отвечает на каждый заказ сообщением OrderPlaced, и тип называет, куда оно уходит: shop-order-events. Очередь это или тема, решает точка монтирования.

Очереди за темой​

src/main.rs
#[subscriber(SqsQueue::new("shop-billing").create_if_missing())]
async fn bill(order: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(order.id, order.amount_cents, "order billed");
    HandlerOutcome::ack()
}

#[subscriber(SqsQueue::new("shop-shipping").create_if_missing())]
async fn ship(order: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

bill и ship - обычные обработчики очередей. Каждая очередь получает собственную копию каждого объявления, поэтому оплата и доставка работают в своём темпе и обрабатывают сообщения заново независимо друг от друга.

Подписка очередей​

src/main.rs
/// Subscribes the billing and shipping queues to the topic, with raw delivery.
///
/// `broker` is a clone of the app's broker, so it uses the connection the app already holds.
async fn subscribe_queues(broker: SqsBroker) -> io::Result<()> {
    let connected = broker.connect().await.map_err(io::Error::other)?;
    for queue in ["shop-billing", "shop-shipping"] {
        connected
            .subscribe_queue_to_topic("shop-order-events", queue)
            .await
            .map_err(io::Error::other)?;
    }
    Ok(())
}

subscribe_queue_to_topic подписывает очередь на тему с прямой доставкой сообщений (raw message delivery). Поэтому копия приходит обычным сообщением SQS: полезная нагрузка - это тело, а каждый заголовок - атрибут сообщения. Без прямой доставки SNS обернул бы сообщение в собственный JSON-конверт. SNS доставляет сообщения в очередь, только если политика доступа очереди разрешает теме отправлять в неё. Перед подпиской вызов добавляет в политику такое правило и сохраняет правила, которые в ней уже есть. Вызов создаёт тему, если её нет, а повторная подписка той же очереди ничего не меняет.

Топология обычно относится к инфраструктуре и создаётся через Terraform или консоль. Здесь её создаёт хук старта на копии брокера приложения, которая использует соединение, уже открытое приложением.

Приложение​

src/main.rs
fn service(broker: SqsBroker) -> RustStream {
    let topology = broker.clone();
    RustStream::new(AppInfo::new("shop", "0.1.0"))
        .after_startup(async move |_state| subscribe_queues(topology).await)
        .with_broker(broker, |b| {
            // The reply goes to the topic named by `OrderPlaced`, through SNS.
            b.include(accept).out_reply(SnsPublish::default());
            b.include(bill);
            b.include(ship);
        })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

.out_reply(SnsPublish::default()) привязывает ответ к SnsPublish, и ответ становится вызовом SNS Publish в тему. Без этого шага тот же ответ уходит через политику брокера по умолчанию, SqsPublish, и попадает в очередь с именем shop-order-events. SnsPublish открывает тему по имени через CreateTopic, который возвращает существующую тему, если она есть, или принимает ARN темы в качестве имени. Та же политика работает в слоте Out и в хуке старта, а ConnectedSqsBroker::sns_publisher даёт издателя вне приложения.

Тема, имя которой оканчивается на .fifo, - это тема FIFO. Она передаёт группу сообщений публикации подписанным на неё очередям FIFO, с теми же шагами группы, которые принимает очередь FIFO. Очередь FIFO подписывается только на тему FIFO.

Для публикации сервису нужны sns:CreateTopic и sns:Publish. Если сервис сам подписывает очереди, ему нужны ещё sns:Subscribe на теме и sqs:GetQueueAttributes и sqs:SetQueueAttributes на каждой очереди.

Тестирование доставки всем подписчикам​

Файл примера запускает одно тело теста дважды. Внутри процесса аккаунт доставляет сообщения темы в очереди, подписанные через subscribe_queue_to_topic, поэтому хук настраивает тест так же, как промышленную среду. Очередь, которую оператор подписал вне сервиса, называется на политике через fans_out_to - шаг фичи testing. Если задана переменная SQS_TEST_ENDPOINT, то же тело выполняется на LocalStack через TestApp::start_live:

cargo test --features sns
SQS_TEST_ENDPOINT=http://localhost:4566 cargo test --features sns -- --test-threads=1

Обвязку и режим работы со стендом разбирает руководство Тестирование сервиса на SQS.