RabbitQuorumQueue

RabbitQuorumQueue consumes a replicated quorum queue, which is always durable and counts the deliveries each message spends.

A quorum queue keeps its messages on several nodes through Raft, so it survives the loss of a node. It is durable, shared and permanent by definition. So the descriptor has no durable, exclusive or auto_delete step, and a quorum queue that contradicts its type does not compile. Every shared setting of RabbitQueue is on it too.

The server counts each delivery of a message, and a retry spends one. A quorum delivery reports its count, where a classic delivery reports only the redelivered flag. Past the queue's delivery limit the server takes the message off the queue: it dead-letters the message when the queue has a dead-letter exchange, and drops it otherwise. RabbitMQ 4 gives a quorum queue a limit of 20 when it names none, so a message that keeps failing gets 21 deliveries. delivery_limit sets another limit; max_attempts with dead_letter at the mount site sets it from the registration.

The queue type is part of the declaration, so a quorum queue is created by an operator or by a broker that declares the topology. A quorum queue refuses some classic arguments: a declaration with x-max-priority fails with AmqpError::Declare.

Example

A reservation waits for stock; one that never fits ends after the queue's default delivery limit.

Macros
Manual

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

/// A reservation waits for stock; the queue gives up on one after its delivery limit.
#[subscriber(RabbitQuorumQueue::new("reservations"))]
async fn reserve(reservation: &Reservation) -> HandlerOutcome {
    if !reservation.in_stock {
        return HandlerOutcome::retry();
    }
    tracing::info!(reservation.id, "reserved");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("reservations", "0.1.0")).with_broker(broker, |b| {
        b.include(reserve);
    })
}

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

/// A reservation waits for stock; the queue gives up on one after its delivery limit.
struct Reserve;

impl Handle<Reservation> for Reserve {
    fn handle(
        &self,
        reservation: &Reservation,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if !reservation.in_stock {
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(reservation.id, "reserved");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("reservations", "0.1.0")).with_broker(broker, |b| {
        let reservations = RabbitQuorumQueue::new("reservations");
        b.include(subscriber(reservations, Reserve).build());
    })
}

Example​

Example