Skip to main content

Exchanges and routing

This guide routes messages through RabbitMQ exchanges: by routing key, to every bound queue, and by headers.

A publisher never writes to a queue. It publishes to an exchange with a routing key, and the exchange copies the message into every queue whose binding matches. In RustStream a queue's bindings are written on its descriptor, and the exchange a message goes to belongs to the publish policy. The routing key is the message's name: the #[outgoing(name = "..")] of its type, or the name the publish gives it.

The example is a shop that announces orders on a topic exchange, price changes on a fanout exchange and shipments on a headers exchange. The broker declares the exchanges, the queues and the bindings, so the example runs on an empty RabbitMQ (Topology).

The default exchange​

Every service so far published to the default exchange. It is a direct exchange that every queue is bound to under its own name. So a message with the routing key orders reaches the queue orders, and a service reads it with #[subscriber("orders")] and no binding at all. A binding on the default exchange is refused, by the server and by the in-process RabbitMQ alike.

A direct exchange of your own routes the same way: a binding names one routing key, and a message reaches the queues bound under exactly its key.

Topic exchanges​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Order {
    id: u64,
    region: String,
}

/// An order event; its routing key is `order.placed.<region>`.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct OrderPlaced {
    id: u64,
    region: String,
}

/// Takes an order from the `orders` queue and announces it on the `events` exchange.
#[subscriber("orders")]
async fn place_order(order: &Order, Out(events): Out<impl Publisher>) -> HandlerOutcome {
    let placed = OrderPlaced {
        id: order.id,
        region: order.region.clone(),
    };
    let key = format!("order.placed.{}", order.region);
    if events.message(&placed).to(&key).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// Every order event of every region.
#[subscriber(RabbitQueue::new("notifications.orders")
    .bind(RabbitExchange::topic("events"), "order.placed.*"))]
async fn notify_customer(event: &OrderPlaced, Ctx(key): Ctx<RoutingKey>) -> HandlerOutcome {
    tracing::info!(event.id, routing_key = %key, "customer notified");
    HandlerOutcome::ack()
}

/// The events of the EU region only.
#[subscriber(RabbitQueue::new("warehouse.eu")
    .bind(RabbitExchange::topic("events"), "*.*.eu"))]
async fn reserve_stock(event: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(event.id, "stock reserved in the EU warehouse");
    HandlerOutcome::ack()
}

/// Every event on the exchange, whatever its key.
#[subscriber(RabbitQueue::new("audit").bind(RabbitExchange::topic("events"), "#"))]
async fn audit(event: &OrderPlaced, Ctx(key): Ctx<RoutingKey>) -> HandlerOutcome {
    tracing::info!(event.id, routing_key = %key, "event audited");
    HandlerOutcome::ack()
}

place_order publishes each order as OrderPlaced with the routing key order.placed.<region>. The type declares no name, so the publish names it with .to(&key). A topic exchange matches the key against each binding's pattern, word by word, with dots between the words. * stands for exactly one word and # for any number of words, none included.

So order.placed.* takes the orders of every region, *.*.eu only the EU's, and # every message on the exchange. An order from the US reaches the notifications and the audit, and the EU warehouse never sees it. A handler reads the key a message arrived with through Ctx<RoutingKey>.

Fanout exchanges​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct PriceChange {
    sku: String,
    price_cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "price.changed")]
struct PriceChanged {
    sku: String,
    price_cents: u64,
}

/// Applies a price change from the catalog and announces it to every reader of `prices`.
#[subscriber("catalog.prices", reply)]
async fn change_price(change: &PriceChange) -> PriceChanged {
    tracing::info!(change.sku, change.price_cents, "price changed");
    PriceChanged {
        sku: change.sku.clone(),
        price_cents: change.price_cents,
    }
}

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

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

A fanout exchange copies every message into every queue bound to it and ignores the routing key. change_price replies with PriceChanged, and the storefront and the search index each get their own copy. The bindings name an empty key, because the exchange reads none. A new reader of price changes is one more queue bound to prices; the publisher does not change.

Headers exchanges​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct PaidOrder {
    id: u64,
    region: String,
    tier: String,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipment")]
struct Shipment {
    order_id: u64,
    region: String,
    tier: String,
}

/// Ships a paid order, with its region and tier as headers, which the exchange routes on.
#[subscriber("orders.paid")]
async fn ship_order(order: &PaidOrder, Out(shipments): Out<impl Publisher>) -> HandlerOutcome {
    let shipment = Shipment {
        order_id: order.id,
        region: order.region.clone(),
        tier: order.tier.clone(),
    };
    let mut attributes = HeaderMap::new();
    attributes.insert("region", order.region.clone());
    attributes.insert("tier", order.tier.clone());
    let sent = shipments
        .message(&shipment)
        .with_headers(attributes)
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The binding arguments: how to match, then the headers matched on.
fn gold_in_eu(x_match: &str) -> FieldTable {
    let mut arguments = FieldTable::default();
    arguments.insert("x-match".into(), AMQPValue::LongString(x_match.into()));
    arguments.insert("region".into(), AMQPValue::LongString("eu".into()));
    arguments.insert("tier".into(), AMQPValue::LongString("gold".into()));
    arguments
}

/// Gold shipments in the EU: both headers have to match.
#[subscriber(RabbitQueue::new("couriers.express")
    .bind_with(RabbitExchange::headers("shipments"), "", gold_in_eu("all")))]
async fn book_express(shipment: &Shipment) -> HandlerOutcome {
    tracing::info!(shipment.order_id, "express courier booked");
    HandlerOutcome::ack()
}

/// Shipments that are gold or in the EU: one header is enough.
#[subscriber(RabbitQueue::new("insurance")
    .bind_with(RabbitExchange::headers("shipments"), "", gold_in_eu("any")))]
async fn insure(shipment: &Shipment) -> HandlerOutcome {
    tracing::info!(shipment.order_id, "shipment insured");
    HandlerOutcome::ack()
}

A headers exchange routes on the message's headers instead of the key. bind_with gives the binding its arguments: the headers to match and x-match, which says how many of them must match. With all, a message needs every one of them; with any, one is enough. ship_order sets region and tier on each shipment with with_headers. A gold shipment in the EU reaches both queues, a silver one in the EU only the insurance, and a silver one in the US neither.

The plain bind is for exchanges that route on the key. On a headers exchange it binds with no headers to match, and all of no headers is true of every message. FieldTable and AMQPValue come from lapin, re-exported in the prelude.

Publishing to an exchange​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        b.include(place_order)
            .out(DefaultSlot, Publish::default().exchange("events"))
            .build();
        b.include(notify_customer);
        b.include(reserve_stock);
        b.include(audit);

        b.include(change_price)
            .out_reply(Publish::default().exchange("prices"));
        b.include(refresh_storefront);
        b.include(reindex);

        b.include(ship_order)
            .out(DefaultSlot, Publish::default().exchange("shipments"))
            .build();
        b.include(book_express);
        b.include(insure);
    })
}

The exchange is named once, on the policy at the mount site: Publish::default().exchange("events"). The handler's body names only the message and its key, so the same handler publishes to another exchange when the mount site says so. A reply takes the same policy at .out_reply(..).

RabbitMQ refuses a publish to an exchange that does not exist and closes the channel the publish went over. Here the subscriptions declare events, prices and shipments at startup, before any handler publishes. A service that publishes to an exchange it does not read from relies on operations or on another service to declare it.

Every binding of a subscription also appears in the service's AsyncAPI document, in the x-ruststream-amqp extension of its channel.

Consistent hashing​

The plugin-consistent-hash feature adds RabbitExchange::consistent_hash, the exchange type of RabbitMQ's consistent hash plugin. It spreads one stream of messages over several queues by a hash of the routing key, and each queue binds with its weight as the routing key. Messages with the same key always reach the same queue. The plugin must be enabled on the server, and the routing happens only there, so its tests run against a live RabbitMQ.

Testing​

The tests in the example file publish an order, a price change and three paid orders, and check which queues each reached. The in-process RabbitMQ routes through the default, direct, topic, fanout and headers exchanges by the same bindings the descriptors declare. Testing a RabbitMQ service covers the harness.