fanout

A fanout exchange copies every message to every bound queue and ignores the routing key.

A queue binds to a fanout exchange with any routing key, usually the empty one. Each bound queue gets its own copy, and each copy is consumed and settled on its own. So a fanout exchange is the broadcast of RabbitMQ: one event, many independent consumers. A service that needs a copy in every instance binds an exclusive queue per instance.

Example

A price change reaches both the search index and the cache, each through a queue of its own.

Macros

examples/lapin_exchanges_fanout.rs

#[subscriber(RabbitQueue::new("prices.search").bind(RabbitExchange::fanout("prices"), ""))]
async fn reindex(price: &PriceChanged) -> HandlerOutcome {
    tracing::info!(price.sku, "search index updated");
    HandlerOutcome::ack()
}

#[subscriber(RabbitQueue::new("prices.cache").bind(RabbitExchange::fanout("prices"), ""))]
async fn recache(price: &PriceChanged) -> HandlerOutcome {
    tracing::info!(price.sku, price.cents, "cache updated");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("catalog", "0.1.0")).with_broker(broker, |b| {
        b.include(publish_price)
            .out(Prices, Publish::default().exchange("prices"))
            .build();
        b.include(reindex);
        b.include(recache);
    })
}

Manual

examples/manual/lapin_exchanges_fanout.rs

struct Reindex;

impl Handle<PriceChanged> for Reindex {
    fn handle(
        &self,
        price: &PriceChanged,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(price.sku, "search index updated");
        ready(Ok(()))
    }
}

struct Recache;

impl Handle<PriceChanged> for Recache {
    fn handle(
        &self,
        price: &PriceChanged,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(price.sku, price.cents, "cache updated");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("catalog", "0.1.0")).with_broker(broker, |b| {
        let prices = RabbitExchange::fanout("prices");
        b.include(subscriber("price-updates", PublishPrice).build())
            .out(Prices, Publish::default().exchange("prices"))
            .build();
        let search = RabbitQueue::new("prices.search").bind(prices.clone(), "");
        b.include(subscriber(search, Reindex).build());
        let cache = RabbitQueue::new("prices.cache").bind(prices, "");
        b.include(subscriber(cache, Recache).build());
    })
}