Struct RabbitQueue 

pub struct RabbitQueue { /* private fields */ }

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

The classic queue is the single-node implementation, and it counts nothing: a delivery says only that it has been seen before, never how often. So a registration’s max_attempts(n) is the runtime’s to apply, counted with the framework’s own header on the copies it publishes, and out_retry(policy) names the publisher those copies leave through. A queue that should end a poison loop on the server instead is a RabbitQuorumQueue.

Descriptors describe the EXPECTED topology for routing; by default nothing is created on the broker (managing infrastructure is the user’s job). Opt in to declaration per broker 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"), "order.*")
    .dead_letter_exchange("dead-letters"))]
async fn on_order(event: &OrderPlaced) -> HandlerOutcome {
    if event.id == 0 {
        // Rejected without requeue: the queue's dead-letter exchange takes it.
        return HandlerOutcome::drop();
    }
    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);
    })
}

A slice parameter asks for a batch, and the mount site says how big it may get:

use std::time::Duration;

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

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

#[subscriber(RabbitQueue::new("settlements")
    .prefetch(nonzero!(64))
    .batch_wait(Duration::from_millis(200)))]
async fn on_settlements(settlements: &[Settlement]) -> HandlerOutcome {
    tracing::info!(count = settlements.len(), "settling");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("settlements", "0.1.0")).with_broker(broker, |b| {
        b.include(on_settlements.batch(nonzero!(32)));
    })
}

The runtime caps the retries of a classic queue and publishes each copy through the out_retry policy, here the confirms one:

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

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

#[subscriber(RabbitQueue::new("refunds"))]
async fn on_refund(refund: &Refund) -> HandlerOutcome {
    tracing::warn!(refund.id, "the provider is down, trying again");
    HandlerOutcome::retry()
}

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("refunds", "0.1.0")).with_broker(broker, |b| {
        b.include(on_refund)
            .max_attempts(nonzero!(5u32))
            .dead_letter("refunds.dead")
            .out_retry(TransactionalPublish::default());
    })
}

Implementations

impl RabbitQueue

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

Describes the classic queue name with the defaults: durable, shared, not auto-deleted.

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

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

RabbitMQ 4 denies transient non-exclusive queues, so a queue declared durable(false) has to be exclusive too.

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

Whether the queue is exclusive to this connection. Defaults to false.

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

Whether the queue is deleted when its last consumer disconnects. Defaults to false.

impl RabbitQueue

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 RabbitQueue

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 RabbitQueue

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

Formats the value using the given formatter.

impl IntoSource for RabbitQueue

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

type Source = RabbitQueue

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl PartialEq for RabbitQueue

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

Equality operator ==.

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

Inequality operator !=.

impl QueueDescriptor for RabbitQueue

impl RedeliveryAddressed<ConnectedLapinBroker> for RabbitQueue

The queue name: on the default exchange a routing key addresses the queue that carries it, so that is where the runtime’s retry copy reaches this subscription again.

The answer holds for a retry publisher on the default exchange, which is what LapinPublish is unless configured otherwise. A publisher aimed at a topic or direct exchange reaches the queue only through a binding under this name, so bind it there or leave the retry publisher on the default exchange.

fn redelivery_address( &self, _connected: &ConnectedLapinBroker, ) -> impl Future<Output = Result<RedeliveryAddress, AmqpError>> + Send

Where a publish reaches this subscription again.

impl StructuralPartialEq for RabbitQueue

impl SubscriptionSource<ConnectedLapinBroker> for RabbitQueue

type Copies = AddressedCopies

The service publishes the copies of a retry, and the queue name is where they go: on the default exchange a routing key addresses the queue that carries it.

A classic queue keeps no counter and has no delivery limit, so there is nothing on the server to end a retry loop; the runtime counts and the mount site names the publisher the copies leave through with out_retry(policy).

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(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

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.