exchange

exchange sends a policy's publishes to a named exchange instead of the default exchange.

The message name stays the routing key, and the exchange routes by it. A topic exchange matches it against binding patterns, a direct exchange matches it exactly, and a fanout exchange ignores it. Without exchange, the default exchange delivers to the queue whose name equals the routing key. RabbitMQ accepts a publish that matches no binding and drops it. Every publish policy of the crate takes the setting: ConfirmsPublish, ServerTxPublish and LapinRequest too. With declare_topology(true), the subscription that binds a queue to the exchange declares the exchange.

Example

Every accepted order is announced on the events topic exchange under order.placed. An audit queue bound under order.* receives it.

Macros
Manual

examples/lapin_publishing_exchange.rs
/// Every accepted order is announced on the `events` exchange under `order.placed`.
#[subscriber(RabbitQueue::new("orders"))]
async fn accept(order: &Order, Out(events): Out<impl Publisher>) -> HandlerOutcome {
    let placed = OrderPlaced { id: order.id };
    if events.message(&placed).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The audit service's queue takes every `order.*` event from that exchange.
#[subscriber(RabbitQueue::new("orders.audit").bind(RabbitExchange::topic("events"), "order.*"))]
async fn audit(placed: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(placed.id, "order placed");
    HandlerOutcome::ack()
}

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

examples/manual/lapin_publishing_exchange.rs
/// Every accepted order is announced on the `events` exchange under `order.placed`.
struct Accept;

impl<Egress> Handle<Order, (), Outs<(Egress,)>> for Accept
where
    Egress: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let placed = OrderPlaced { id: order.id };
        let events = outs.get(DefaultSlot);
        if events.message(&placed).publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

/// The audit service's queue takes every `order.*` event from that exchange.
struct Audit;

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

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let audit =
            RabbitQueue::new("orders.audit").bind(RabbitExchange::topic("events"), "order.*");
        b.include(subscriber(RabbitQueue::new("orders"), Accept).build())
            .out(DefaultSlot, Publish::default().exchange("events"))
            .build();
        b.include(subscriber(audit, Audit).build());
    })
}

Example​

Example