declare_topology

declare_topology(true) declares the exchanges, the queue and the bindings a descriptor states; off, the topology is the operator's and must exist.

The setting is off by default. A descriptor then states the topology the service expects, and the service creates nothing. A subscription to a queue the server does not have fails startup with AmqpError::Subscribe.

With the setting on, each subscription declares its topology before it starts consuming. It declares the exchanges its bindings name, then the queue, then the bindings, in that order. The built-in amq.* exchanges and the default exchange are never declared: the server already has them. A declaration repeats safely while the descriptor matches what the server holds. A queue or an exchange declared again with other settings is refused by the server with PRECONDITION_FAILED, and startup fails with AmqpError::Declare. RabbitMQ 4 refuses a transient queue that is not exclusive, so a RabbitQueue declared durable(false) has to be exclusive(true) too.

Some settings exist only as declaration arguments, so they reach the server only from a service that declares. A quorum queue's delivery limit and the waiting queue of a delayed redelivery are two of them.

The in-process transport finds every queue a subscription names, so a test shows no missing queue. A declaration the server refuses is refused in process too.

Example

The service declares the topic exchange events, the queue orders and the binding order.* between them.

Macros
Manual

examples/lapin_broker_declare_topology.rs
use ruststream_lapin::prelude::*;

#[subscriber(RabbitQueue::new("orders").bind(RabbitExchange::topic("events"), "order.*"))]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The service owns `events`, `orders` and the binding between them.
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
    })
}

examples/manual/lapin_broker_declare_topology.rs
use ruststream_lapin::prelude::*;

struct Place;

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

fn app() -> impl App {
    // The service owns `events`, `orders` and the binding between them.
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let orders = RabbitQueue::new("orders").bind(RabbitExchange::topic("events"), "order.*");
        b.include(subscriber(orders, Place).build());
    })
}

Example​

Example