Skip to main content

Topology

This guide describes a service's queues, exchanges and bindings, and decides who creates them: the service or the people who operate RabbitMQ.

A queue descriptor states what the service expects to find on the server. Nothing is created unless the broker opts in with declare_topology(true). The example is a payments service with a replicated queue for payments, a queue for the payments that fail, and a queue of each instance's own for exchange rates.

Classic and quorum queues​

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

/// Payments live in a replicated quorum queue; what it rejects goes to `payments.dead`.
#[subscriber(RabbitQuorumQueue::new("payments").dead_letter_exchange("payments.dead"))]
async fn charge(payment: &Payment) -> HandlerOutcome {
    if payment.card.is_empty() {
        tracing::warn!(payment.id, "payment without a card rejected");
        return HandlerOutcome::drop();
    }
    tracing::info!(payment.id, payment.amount_cents, "payment charged");
    HandlerOutcome::ack()
}

/// Every payment the `payments` queue dead-letters.
#[subscriber(RabbitQueue::new("payments.failed")
    .bind(RabbitExchange::fanout("payments.dead"), ""))]
async fn record_failure(payment: &Payment) -> HandlerOutcome {
    tracing::error!(payment.id, "payment failed");
    HandlerOutcome::ack()
}

RabbitMQ has two queue types, and the type is fixed when the queue is created. A classic queue lives on one node. A quorum queue is replicated over the cluster by Raft and survives the loss of a node. Each type has its own descriptor: RabbitQueue and RabbitQuorumQueue. Take a quorum queue for data that must not be lost, such as payments, and a classic queue for the rest. RabbitMQ's quorum queue guide compares the two.

A quorum queue is always durable, shared between connections and permanent. So RabbitQuorumQueue has no durable, exclusive or auto_delete, and a quorum queue that contradicts its type does not compile. A quorum queue also counts the deliveries of each message, which the retries guide builds on.

dead_letter_exchange gives payments a route for what it rejects. A payment charge drops goes to the payments.dead exchange, and from there to payments.failed, the queue bound to it. dead_letter_routing_key replaces the message's routing key on that route when the exchange routes by key.

A queue of each instance​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct FxRate {
    currency: String,
    micros_per_euro: u64,
}

/// The name of this instance's own queue.
fn rates_queue_name() -> String {
    format!("payments.rates.{}", process::id())
}

/// A queue of this instance's own: it lives as long as the connection, and a rate older than a
/// minute expires unread.
#[subscriber(RabbitQueue::new(rates_queue_name())
    .durable(false)
    .exclusive(true)
    .auto_delete(true)
    .argument("x-message-ttl", AMQPValue::LongLongInt(60_000))
    .bind(RabbitExchange::fanout("fx-rates"), ""))]
async fn update_rate(rate: &FxRate) -> HandlerOutcome {
    tracing::info!(rate.currency, rate.micros_per_euro, "exchange rate updated");
    HandlerOutcome::ack()
}

Every instance needs every exchange rate, so each instance reads a queue of its own, bound to the fx-rates fanout exchange. The queue name carries the process id, so two instances never share it. exclusive(true) lets only this connection use the queue, and RabbitMQ deletes it when the connection closes. auto_delete(true) deletes it once its last consumer is gone. durable(false) keeps it off the disk: rates are worthless after a restart. RabbitMQ 4 accepts a transient queue only when it is exclusive, so durable(false) comes with exclusive(true).

argument passes a declaration argument the descriptor has no method for, here x-message-ttl: a rate nobody read within a minute expires. The same way sets x-max-length, x-max-priority, x-overflow or x-single-active-consumer. The value is an AMQPValue, and arguments takes a whole table.

Declared by the service​

src/main.rs
fn service(broker: LapinBroker) -> RustStream {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(charge);
        b.include(record_failure);
        b.include(update_rate);
    })
}

#[ruststream::app]
fn app() -> impl App {
    // Development declares the topology; production runs with `DECLARE_TOPOLOGY=false` against
    // the queues operations created.
    let declare = env::var("DECLARE_TOPOLOGY").map_or(true, |value| value != "false");
    service(LapinBroker::new("amqp://localhost:5672").declare_topology(declare))
}

With declare_topology(true), each subscription declares at startup what its descriptor states. It declares the exchanges its bindings name first, then the queue with its arguments, then the bindings. The default exchange and the built-in amq.* exchanges exist on every server and are never declared. A declaration repeats harmlessly while it matches what the server has, so every instance of the service declares the same queues.

A declaration that differs from the existing queue fails. RabbitMQ answers PRECONDITION_FAILED to a queue declared again with another durability or other arguments, and the service does not start: AmqpError::Declare names the queue. A queue's type and arguments change by deleting and declaring it again, which loses what it holds, so they change through a migration, not a deploy. RabbitMQ explains which properties must match in property equivalence.

Owned by operations​

DECLARE_TOPOLOGY=false starts the same service with declaration off. The descriptors then declare nothing: they state the queues, arguments and bindings the service expects, and operations create them. A queue that does not exist fails the subscription at startup with AmqpError::Subscribe, so a missing queue stops the deploy instead of losing messages.

Operations often set limits and TTLs with a RabbitMQ policy instead of declaration arguments. A policy changes a queue without declaring it again, and an argument set by a policy never conflicts with a service's declaration. The descriptor then names only what the service itself relies on.

Declaration off also changes what a quorum queue's retry declaration can do, which the retries guide describes.

Testing​

The tests in the example file reject a payment and find it in payments.failed, under declaration on and off. The in-process RabbitMQ follows the broker's setting: it declares what the descriptors state and refuses what the server refuses, such as a transient queue that is not exclusive, a binding on the default exchange or a queue declared again with other settings. With declaration off, every queue a subscription names is there in process, so a missing queue belongs to a live test. Testing a RabbitMQ service covers the harness.