RabbitExchange

bind joins a queue to a RabbitExchange under a routing key; a direct exchange routes on the exact key, and the default exchange on queue names.

A publisher sends a message to an exchange, never to a queue. The exchange copies the message into the queues bound to it whose bindings match. RabbitExchange names the exchange of a binding, and its constructor names its type:

Constructor	Routes on
`direct`	the exact routing key
`topic`	a dotted pattern of the routing key
`fanout`	nothing: every bound queue gets every message
`headers`	the message's headers, against the binding's arguments
`consistent_hash`	a hash of the routing key, through a plugin

bind(exchange, routing_key) adds a binding to a queue descriptor, and a queue may carry several. A direct exchange delivers a message to every queue bound under its routing key exactly; a message under a key no queue is bound under goes nowhere. The default exchange, with the empty name, is a direct exchange every queue is bound to under its own name. So a message published with no exchange reaches the queue its routing key names, and a binding to the default exchange is refused.

An exchange is durable by default and survives a broker restart, and auto_delete is off. The exchange and its bindings reach the server when the broker declares the topology; the built-in amq.* exchanges are bound to and never declared. The publishing side names its exchange on the publish policy with exchange, and the message's name is its routing key.

RabbitExchange::custom(name, kind) names an exchange of a type a server plugin provides, such as x-random or x-modulus-hash. A queue binds to it with bind, or with bind_with for a plugin that routes on binding arguments. The plugin does the routing, so the server needs the plugin enabled, and the in-process transport routes nothing through such an exchange.

Example

Checkout publishes each payment to the direct exchange payments under its method; the card queue is bound under card, the invoice queue under invoice.

Macros
Manual

examples/lapin_exchanges_direct.rs
#[subscriber(RabbitQueue::new("payments.card").bind(RabbitExchange::direct("payments"), "card"))]
async fn charge_card(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.order_id, "card charged");
    HandlerOutcome::ack()
}

#[subscriber(RabbitQueue::new("payments.invoice")
    .bind(RabbitExchange::direct("payments"), "invoice"))]
async fn send_invoice(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.order_id, "invoice sent");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(checkout)
            .out(Payments, Publish::default().exchange("payments"))
            .build();
        b.include(charge_card);
        b.include(send_invoice);
    })
}

examples/manual/lapin_exchanges_direct.rs
struct ChargeCard;

impl Handle<Payment> for ChargeCard {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payment.order_id, "card charged");
        ready(Ok(()))
    }
}

struct SendInvoice;

impl Handle<Payment> for SendInvoice {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payment.order_id, "invoice sent");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        let payments = RabbitExchange::direct("payments");
        b.include(subscriber("checkout", Checkout).build())
            .out(Payments, Publish::default().exchange("payments"))
            .build();
        let card = RabbitQueue::new("payments.card").bind(payments.clone(), "card");
        b.include(subscriber(card, ChargeCard).build());
        let invoice = RabbitQueue::new("payments.invoice").bind(payments, "invoice");
        b.include(subscriber(invoice, SendInvoice).build());
    })
}

Example​

Example