Struct RabbitQuorumQueue 

pub struct RabbitQuorumQueue { /* private fields */ }

Describes one quorum-queue subscription: the queue, its expected settings, and its bindings.

A quorum queue counts the deliveries a message has spent and ends a poison loop itself: with an x-delivery-limit and a dead-letter route it carries a message away once its deliveries run out, whether or not a service is running to count them. So the retries of a registration mounted here are the broker’s, and out_retry(policy) does not compile on it: the server moves the spent message, and the service publishes no copy. A registration declares the pair at the mount site:

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

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

#[subscriber(RabbitQuorumQueue::new("payouts"))]
async fn on_payout(payout: &Payout) -> HandlerOutcome {
    tracing::warn!(payout.id, "the bank refused, trying again");
    HandlerOutcome::retry()
}

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("payouts", "0.1.0")).with_broker(broker, |b| {
        // The server applies this cap: it becomes the queue's delivery limit and dead-letter
        // route.
        b.include(on_payout)
            .max_attempts(nonzero!(5u32))
            .dead_letter("payouts.dead");
    })
}

The pair becomes this queue’s x-delivery-limit and dead-letter route when the broker opts into declare_topology(true). A queue declared elsewhere carries its own arguments, so subscribing then refuses the declaration rather than promising a cap nothing applies; configure the queue on the broker and mount it without the pair. The crate reads the queue’s count either way.

A quorum queue is durable, shared and permanent by definition, so those three settings are not here to be set wrong. For the single-node implementation take RabbitQueue.

Examples

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

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

// A charge the payment provider keeps refusing is moved to `dead-letters` by the server after
// its fifth delivery: the limit counts the four returns it survives.
#[subscriber(RabbitQuorumQueue::new("charges")
    .bind(RabbitExchange::topic("events"), "charge.*")
    .dead_letter_exchange("dead-letters")
    .delivery_limit(4))]
async fn take_payment(charge: &Charge) -> HandlerOutcome {
    tracing::info!(charge.id, "charging");
    HandlerOutcome::retry()
}

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

Implementations

impl RabbitQuorumQueue

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

Describes the quorum queue name.

pub fn delivery_limit(self, returns: u32) -> Self

How often the queue returns a message before it carries it away (x-delivery-limit).

The argument counts the returns a message survives, so it is one less than the number of deliveries the queue allows: delivery_limit(2) hands a message out three times. Where it goes then is dead_letter_exchange; a limit with no dead-letter route drops the spent message instead.

This is the queue’s own policy, for a queue whose retries are not one registration’s business. A mount site that declares max_attempts(n).dead_letter(..) is the more specific statement and is written over this.

impl RabbitQuorumQueue

pub fn bind( self, exchange: RabbitExchange, routing_key: impl Into<String>, ) -> Self

Binds the queue to exchange under routing_key.

Call repeatedly for multiple bindings. Without any binding the queue only receives messages published to the default exchange under the queue name.

The binding carries no arguments, which is what every exchange that routes on the routing key wants. A headers exchange routes on arguments instead, and an empty table there matches every message it gets rather than none, so bind to one with bind_with.

The generated AsyncAPI document reports every binding in the crate’s own x-ruststream-amqp extension, in declaration order; the specification’s binding has no field for one.

pub fn bind_with( self, exchange: RabbitExchange, routing_key: impl Into<String>, arguments: FieldTable, ) -> Self

Binds the queue to exchange under routing_key, with arguments for the exchange to match on.

A headers exchange routes by these rather than by the routing key: x-match decides whether a message has to carry all of the remaining entries or any one of them, and the routing key is ignored (pass ""). Binding to one without arguments quietly takes the whole exchange - all of no headers is true of every message - so the plain bind belongs to the exchanges that route on the key.

The crate overview has the worked all and any example; AMQPValue and FieldTable are re-exported for building the table. The generated AsyncAPI document reports the arguments with the binding, in the crate’s own x-ruststream-amqp extension.

pub fn dead_letter_exchange(self, exchange: impl Into<String>) -> Self

Dead-letters rejected messages to exchange (the x-dead-letter-exchange argument).

A handler returning drop settles with basic.reject(requeue = false), which routes the message there. This is the queue’s own topology, and it applies to every rejection, whoever caused it.

pub fn dead_letter_routing_key(self, routing_key: impl Into<String>) -> Self

Overrides the routing key dead-lettered messages carry (x-dead-letter-routing-key).

pub fn argument(self, name: impl Into<String>, value: AMQPValue) -> Self

Sets one raw declaration argument (x-...), passed through verbatim.

Panics

Panics if name exceeds 255 bytes (the AMQP short-string limit); argument names are compile-time constants in practice.

pub fn arguments(self, arguments: FieldTable) -> Self

Replaces the whole raw declaration argument table (x-... passthrough).

pub fn prefetch(self, prefetch: NonZeroU16) -> Self

Caps unacknowledged deliveries in flight for this subscription (basic.qos), overriding the broker-wide prefetch.

This is the back-pressure window for the subscriber stream. When neither is set the server imposes no prefetch limit.

The count is a NonZeroU16 because AMQP reads basic.qos(0) as “no limit”, the exact opposite of a cap: the zero sentinel is unrepresentable here, and leaving the prefetch unset is how “unlimited” is spelled.

pub fn batch_wait(self, batch_wait: Duration) -> Self

Caps how long a partial batch waits for more deliveries before it goes to the handler. Defaults to 50 ms.

AMQP delivers one message at a time, so a batch handler’s batch(n) is honoured by collecting deliveries on the client; how big a batch may be is the registration’s word, and this is the other half of the trade-off - the ceiling on how long a batch that never fills keeps its deliveries. Raise it on a slow link or a sparse queue where fuller batches are worth the wait; lower it where a batch arriving late costs more than a batch arriving short. It has no effect on a subscription without a batch handler.

pub fn delay(self, delay: Delay) -> Self

Makes retry_after / nack_after native, routing delayed redeliveries through a broker waiting queue instead of the core in-process fallback.

Without this the runtime handles a delay with its broker-agnostic deferred re-publish (at-most-once over the delay window, held in the service). With it, a delayed message is re-published to the Delay waiting queue with a per-message TTL and dead-lettered back to this queue when it fires, so the delayed copy lives on the broker.

The copy the waiting queue releases is a new message, which the server counts from zero; the framework’s own retry count is what survives the round.

The waiting queue is infrastructure: it is declared only when the broker opts into declare_topology; otherwise provision it yourself. Without declaration the subscription checks that it exists when it opens, and refuses to open when it does not, because a delayed copy sent to a missing waiting queue would be dropped.

pub fn name(&self) -> &str

The queue name.

Trait Implementations

impl Clone for RabbitQuorumQueue

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 RabbitQuorumQueue

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

Formats the value using the given formatter.

impl IntoSource for RabbitQuorumQueue

The descriptor is its own source, so the manual path’s subscriber(source, body) takes a RabbitQuorumQueue where the attribute path writes it in the decorator.

type Source = RabbitQuorumQueue

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl PartialEq for RabbitQuorumQueue

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

Equality operator ==.

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

Inequality operator !=.

impl QueueDescriptor for RabbitQuorumQueue

impl StructuralPartialEq for RabbitQuorumQueue

impl SubscriptionSource<ConnectedLapinBroker> for RabbitQuorumQueue

type Copies = BrokerMoves

The queue moves a spent delivery itself: its x-delivery-limit counts the deliveries and its dead-letter route takes the message that runs out of them, with nothing published by this process. That is what makes out_retry(policy) a compile error here, and what keeps the cap holding while the service is down.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

Takes the registration’s cap and dead-letter destination into the descriptor, so the queue this broker declares carries them as topology.

fn channel_bindings(&self) -> Bindings

The queue this subscription consumes, as the amqp binding names its settings.

fn operation_bindings(&self) -> Bindings

The consumer’s own half: this crate acknowledges by hand on every subscription.

type Subscriber = LapinSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedLapinBroker, ) -> Result<Self::Subscriber, AmqpError>

Opens the subscription against the connected broker. Called once at startup.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.