Skip to main content

Fan-out with SNS

This guide publishes a message to an SNS topic once and lets SNS deliver a copy to every SQS queue subscribed to the topic.

SNS to SQS is how AWS services fan out: one topic, one queue per consumer. The crate publishes to topics; each queue behind a topic is read the ordinary way. Publishing to SNS needs the sns feature:

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

The example is a shop. It accepts an order and announces it on a topic, and billing and shipping each read the announcement from a queue of their own.

A reply to a topic​

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 answers each order with an OrderPlaced, and the type names where it goes: shop-order-events. The mount site decides whether that name is a queue or a topic.

The queues behind the topic​

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 and ship are ordinary queue handlers. Each queue gets its own copy of every announcement, so billing and shipping work at their own pace and retry on their own.

Subscribing the queues​

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 subscribes a queue to a topic with raw message delivery. So a copy arrives as a plain SQS message: the payload is the body, and each header is a message attribute. Without raw delivery SNS would wrap the message in a JSON envelope of its own. SNS delivers to a queue only when the queue's access policy lets the topic send to it. The call adds that statement to the policy before it subscribes, and keeps the statements already there. The call creates the topic when it does not exist, and subscribing the same queue again changes nothing.

Topology is usually infrastructure, created with Terraform or the console. Here a startup hook does it on a clone of the app's broker, which uses the connection the app already holds.

The app​

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()) binds the reply to SnsPublish, and the reply becomes an SNS Publish to the topic. Without that step the same reply rides the broker's default policy, SqsPublish, and lands on a queue named shop-order-events. SnsPublish opens a topic by name with CreateTopic, which returns the existing topic when there is one, or takes the topic's ARN as the name. The same policy works in an Out slot and in a startup hook, and ConnectedSqsBroker::sns_publisher gives a publisher outside the app.

A topic whose name ends in .fifo is a FIFO topic. It carries the message group of a publish to the FIFO queues subscribed to it, with the group steps a FIFO queue takes. A FIFO queue subscribes only to a FIFO topic.

The service needs sns:CreateTopic and sns:Publish to publish. To subscribe queues itself it also needs sns:Subscribe on the topic and sqs:GetQueueAttributes and sqs:SetQueueAttributes on each queue.

Testing the fan-out​

The example file runs one test body twice. In process, the in-process account delivers a topic's messages to the queues subscribed with subscribe_queue_to_topic, so the hook wires the test as it wires production. A queue an operator subscribed outside the service is named on the policy with fans_out_to, a step of the testing feature. With SQS_TEST_ENDPOINT set, the same body runs against LocalStack through TestApp::start_live:

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

Testing an SQS service covers the harness and the live mode.