topic

AmqpAddress::topic asks for the topic terminus: every subscription on the address gets its own copy of each message.

The descriptor attaches its receiver with the terminus capability topic on the link's source. ActiveMQ Artemis and the other brokers that read the capability treat the address as multicast. That is a broadcast: an audit trail and a notifier read the same events, each at its own pace. A deferred retry publishes its copy to the address, so every topic subscription on it receives the copy, not only the one that asked. In process, TestApp copies each message to every topic subscription of the address.

Example

Two subscriptions of one service on the address orders.placed.

Macros

examples/amqp_subscriptions_topic.rs

/// Audit keeps a record of every placed order.
#[subscriber(AmqpAddress::topic("orders.placed"))]
async fn audit(event: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(event.id, "order recorded");
    HandlerOutcome::ack()
}

/// Notifications tell the customer about the same orders, from a copy of their own.
#[subscriber(AmqpAddress::topic("orders.placed"))]
async fn notify(event: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(event.id, "customer notified");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(audit);
        b.include(notify);
    })
}

Manual

examples/manual/amqp_subscriptions_topic.rs

/// Audit keeps a record of every placed order.
struct Audit;

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

/// Notifications tell the customer about the same orders, from a copy of their own.
struct Notify;

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

fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(AmqpAddress::topic("orders.placed"), Audit).build());
        b.include(subscriber(AmqpAddress::topic("orders.placed"), Notify).build());
    })
}