SnsPublish

SnsPublish fans a reply out from an SNS topic to every queue subscribed to it; one out_reply step switches a position to it.

The policy needs the crate's sns feature. A mount site names it where a registration publishes: .out_reply(SnsPublish::default()) for a reply, .out(Slot, SnsPublish::default()) for a slot. The destination reads the same as on SqsPublish: the reply type or the mount site names it, and the policy decides whether the name is a queue or a topic. Without the step the same reply lands on a queue of that name. The runtime pairs the policy with the connected broker into an SnsPublisher.

The destination is a topic name or a topic ARN. A name resolves to its ARN through CreateTopic, which returns an existing topic as it is and creates a missing one. A name ending in .fifo is a FIFO topic, and it takes the same group and deduplication steps as a FIFO queue. A topic and a queue may share a name: each publish goes where its policy sends it. The payload is the notification's message, and the headers travel as message attributes. SNS takes a message of up to 256 KiB.

SNS is a publish target only: the service reads the queues subscribed to the topic. A queue joins a topic through subscribe_queue_to_topic, or through the operator's own provisioning. Either way, SNS delivers to the queue only when the queue's access policy lets the topic send to it, and subscribe_queue_to_topic adds that statement itself.

Example

The dependency line:

ruststream-sqs-sns = { version = "0.7", features = ["sns"] }

The notification names its topic, so the handler answers with a bare reply clause.

Macros
Manual

examples/sns_fanout.rs
/// The notification the topic fans out. It names the topic, so the handler answers with a bare
/// `reply` clause.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders-events")]
struct OrderPlaced {
    id: u64,
}

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

examples/manual/sns_fanout.rs
/// The notification the topic fans out. It names the topic, so the reply takes no destination
/// at the mount site.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[outgoing(name = "orders-events")]
struct OrderPlaced {
    id: u64,
}

struct Accept;

impl Handle<PlaceOrder, OrderPlaced> for Accept {
    fn handle(
        &self,
        order: &PlaceOrder,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<OrderPlaced, HandlerOutcome>> {
        tracing::info!(order.id, "order accepted");
        ready(Ok(OrderPlaced { id: order.id }))
    }
}

The mount site binds the reply position to SnsPublish, and billing and shipping each receive the order.

Macros
Manual

examples/sns_fanout.rs
fn service(broker: SqsBroker) -> impl App {
    // A clone connects through the connection the app's broker holds.
    let topology = broker.clone();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .after_startup(async move |_state| wire_topology(topology).await)
        .with_broker(broker, |b| {
            // One step moves the reply from the queue of that name to the topic.
            b.include(accept).out_reply(SnsPublish::default());
            b.include(bill);
            b.include(ship);
        })
}

examples/manual/sns_fanout.rs
fn service(broker: SqsBroker) -> impl App {
    // A clone connects through the connection the app's broker holds.
    let topology = broker.clone();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .after_startup(async move |_state| wire_topology(topology).await)
        .with_broker(broker, |b| {
            let accept = subscriber("orders", Accept)
                .map_source(|_| SqsQueue::new("orders").create_if_missing());
            // One step moves the reply from the queue of that name to the topic.
            b.include(accept.reply().build())
                .out_reply(SnsPublish::default());
            for queue in ["billing", "shipping"] {
                let record = subscriber(queue, Record(queue))
                    .map_source(move |_| SqsQueue::new(queue).create_if_missing());
                b.include(record.build());
            }
        })
}

Example​

Example