Struct RabbitExchange 

pub struct RabbitExchange { /* private fields */ }

Describes the exchange side of a RabbitQueue binding.

Like every descriptor in this crate it only records the EXPECTED topology; nothing is declared unless the broker was built with declare_topology(true).

Examples

use ruststream_lapin::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct OrderPlaced {
    id: u64,
}

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

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(on_order);
    })
}

Implementations

impl RabbitExchange

pub fn direct(name: impl Into<String>) -> Self

A direct exchange: routes on an exact routing-key match.

pub fn topic(name: impl Into<String>) -> Self

A topic exchange: routes on dot-separated routing-key patterns (order.*).

pub fn fanout(name: impl Into<String>) -> Self

A fanout exchange: routes every message to every bound queue.

pub fn headers(name: impl Into<String>) -> Self

A headers exchange: routes on the binding’s arguments instead of the routing key.

A queue binds to one with bind_with, whose argument table carries x-match (all or any) and the headers a message has to match. The plain bind sends no arguments, and an empty table here matches every message the exchange gets rather than none.

Examples

use ruststream_lapin::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// 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
}

#[subscriber(RabbitQueue::new("orders.eu.gold")
    .bind_with(RabbitExchange::headers("orders"), "", gold_in_eu("all")))]
async fn on_gold_in_eu(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "gold order in the EU");
    HandlerOutcome::ack()
}

// One header is enough here, where the binding above needs both.
#[subscriber(RabbitQueue::new("orders.eu.or.gold")
    .bind_with(RabbitExchange::headers("orders"), "", gold_in_eu("any")))]
async fn on_gold_or_eu(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "gold order or an order in the EU");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(on_gold_in_eu);
        b.include(on_gold_or_eu);
    })
}

pub fn custom(name: impl Into<String>, kind: impl Into<String>) -> Self

An exchange of a plugin-provided type, for example "x-delayed-message".

A plugin that routes by binding arguments takes them through bind_with.

pub fn consistent_hash(name: impl Into<String>) -> Self

A consistent-hash exchange (the x-consistent-hash type from the rabbitmq_consistent_hash_exchange plugin): distributes messages across the queues bound to it by hashing the routing key, so partition-like fan-out is a server-side concern rather than a client one.

Each queue binds with a routing key that is its integer weight ("1", "2", …); the broker splits the hash space proportionally. Requires the plugin to be enabled on the broker, so it lives behind the plugin-consistent-hash feature.

Examples

use ruststream_lapin::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// This queue binds with weight 1, so it takes its share of the hash space.
#[subscriber(RabbitQueue::new("orders.shard-a")
    .bind(RabbitExchange::consistent_hash("orders-by-key"), "1"))]
async fn on_order(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "shard a took the order");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(on_order);
    })
}

pub fn durable(self, durable: bool) -> Self

Whether the exchange survives a broker restart. Defaults to true.

pub fn auto_delete(self, auto_delete: bool) -> Self

Whether the exchange is deleted when its last binding is removed. Defaults to false.

pub fn name(&self) -> &str

The exchange name.

Trait Implementations

impl Clone for RabbitExchange

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for RabbitExchange

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Eq for RabbitExchange

impl PartialEq for RabbitExchange

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for RabbitExchange