topic

A topic exchange routes on dotted routing-key patterns, where * matches one word and # any number of words.

A routing key of a topic exchange is a list of words separated by dots, such as order.created or order.shipped.late. A queue binds with a pattern of the same shape. * stands for exactly one word, and # for zero or more words. So order.* takes order.created and passes order.shipped.late by, and order.# takes both. A pattern without wildcards matches its key exactly, as on a direct exchange. A message that matches several bindings of one queue reaches that queue once.

A handler that serves several keys reads the key of each delivery through the RoutingKey context key.

Example

Order steps go to the topic exchange events as order.<step>; the audit queue takes every one-word step, and the shipping queue takes order.shipped alone.

Macros
Manual

examples/lapin_exchanges_topic.rs
/// Every order event: `order.created`, `order.shipped` and the rest.
#[subscriber(RabbitQueue::new("audit").bind(RabbitExchange::topic("events"), "order.*"))]
async fn audit(event: &OrderEvent, Ctx(key): Ctx<RoutingKey>) -> HandlerOutcome {
    tracing::info!(event.id, %key, "order event audited");
    HandlerOutcome::ack()
}

/// Only the shipments.
#[subscriber(RabbitQueue::new("shipping")
    .bind(RabbitExchange::topic("events"), "order.shipped"))]
async fn notify_shipping(event: &OrderEvent) -> HandlerOutcome {
    tracing::info!(event.id, "shipping notified");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(announce)
            .out(Events, Publish::default().exchange("events"))
            .build();
        b.include(audit);
        b.include(notify_shipping);
    })
}

examples/manual/lapin_exchanges_topic.rs
/// Every order event: `order.created`, `order.shipped` and the rest.
struct Audit;

impl Handle<OrderEvent, (), (), AmqpContext> for Audit {
    fn handle(
        &self,
        event: &OrderEvent,
        _outs: &(),
        ctx: &mut Context<'_, AmqpContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let key = ctx.context(RoutingKey);
        tracing::info!(event.id, %key, "order event audited");
        ready(Ok(()))
    }
}

/// Only the shipments.
struct NotifyShipping;

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

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let events = RabbitExchange::topic("events");
        b.include(subscriber("order-steps", Announce).build())
            .out(Events, Publish::default().exchange("events"))
            .build();
        let audit = RabbitQueue::new("audit").bind(events.clone(), "order.*");
        b.include(subscriber(audit, Audit).build());
        let shipping = RabbitQueue::new("shipping").bind(events, "order.shipped");
        b.include(subscriber(shipping, NotifyShipping).build());
    })
}

Example​

Example