Struct RabbitMqOptions 

pub struct RabbitMqOptions {
    pub connection_properties: ConnectionProperties,
    pub dead_suffix: String,
    pub declare_dead_letter_queues: bool,
    pub deferred_suffix: String,
    pub retry_granularity: Duration,
    pub deferred_granularity: Duration,
    pub reconnect: Option<Arc<dyn ReconnectPolicy>>,
}

Configuration for RabbitMqBackend::with_options.

use queuey_rabbitmq::RabbitMqOptions;

let options = RabbitMqOptions::default()
    .dead_suffix("-dlq")
    .declare_dead_letter_queues(false);
assert_eq!(options.dead_suffix, "-dlq");

Fields

connection_properties: ConnectionProperties

Handshake properties passed to lapin::Connection::connect.

dead_suffix: String

Suffix appended to a queue name to name its dead-letter queue.

Defaults to DEFAULT_DEAD_SUFFIX (".dead").

declare_dead_letter_queues: bool

Whether declare also declares the q.dead queues.

Defaults to true. Set to false when dead-letter queues are managed out of band (policies, an operator-owned topology, a different broker vhost).

This flag also selects how a message is dead-lettered, because this backend never publishes to a queue it does not own:

• true: Delivery::dead_letter publishes the envelope to q.dead with the x-death-* headers and then acks the original.
• false: nothing is published. The original is rejected with requeue = false, so the broker applies whatever x-dead-letter-exchange policy the operator put on q, and drops the message if there is none. The reason is logged at WARN, since it is not recorded anywhere else.

The same choice governs a message whose body is not a valid envelope.

deferred_suffix: String

Infix between a queue name and a hold queue’s TTL.

Defaults to DEFAULT_DEFERRED_SUFFIX (".deferred"), so a 30-second wait on myapp.emails happens in myapp.emails.deferred.30000. Retries, delayed enqueues and deferrals all wait in these hold queues; see crate::topology for why the TTL is part of the name.

retry_granularity: Duration

Step that retry backoffs and Producer::enqueue_after delays are rounded up to.

Defaults to one second. Every distinct rounded delay gets its own hold queue, so this is the knob that trades backoff precision for the number of queues on the broker. It matters most for exponential backoff with jitter, which produces a different delay for every retry: with the default, a policy capped at five minutes can create at most 300 hold queues per work queue, and a granularity of ten seconds brings that down to 30. Idle hold queues delete themselves, so this bounds the number that exist at once, not a total.

Separate from deferred_granularity on purpose: a backoff is a heuristic that tolerates coarse rounding, a Retry-After is a contract that may not.

A retry is never released early: rounding is always up, a delay shorter than the granularity still waits one full step, and a delay that rounds up past MAX_DEFERRAL_MS (~24.8 days) is refused instead of being shortened. A zero (or sub-millisecond) value is clamped to one millisecond rather than rejected, exactly as for deferred_granularity.

deferred_granularity: Duration

Step that deferral delays are rounded up to.

Defaults to one second. Every distinct rounded delay gets its own hold queue, so this is the knob that trades precision for the number of queues on the broker: with the default, Retry-After: 30 and a computed 29.2s delay share q.deferred.30000, and no deferral can create more than MAX_TTL_MS / 1000 queues per work queue.

A deferral is never released early: rounding is always up, a delay shorter than the granularity still waits one full step, and a delay that rounds up past MAX_DEFERRAL_MS (~24.8 days) is refused instead of being shortened.

A zero (or sub-millisecond) value is clamped to one millisecond by deferred_ttl_ms rather than rejected, because a backend constructor must not panic on a config value, but one millisecond of granularity means up to one hold queue per distinct millisecond, which is almost never what you want.

Note what is not here: nothing tunes a hold queue’s x-expires. Its arguments are a pure function of its name (x-expires = 2 * ttl), so two processes configured differently still agree on q.deferred.30000 instead of locking each other out with PRECONDITION_FAILED. Both granularities are safe to tune because they only change which hold queue a delay lands in, never that queue’s arguments.

reconnect: Option<Arc<dyn ReconnectPolicy>>

How a lost connection is recovered, or None to fail instead.

Defaults to BackoffPolicy::default: retry forever with a jittered exponential backoff. Any ReconnectPolicy implementation can go here, so pacing that a backoff curve cannot express (a circuit breaker, a schedule, a different answer for an authentication failure than for a refused connection) is a matter of writing one. A dropped connection is then invisible to job code and to Worker::run, which keeps running: publishes wait for the connection to come back, and consumers resubscribe on it.

Two consequences worth knowing before relying on it:

• A broker that never comes back looks like a stall, not an error. With the default unlimited policy nothing ever returns Err; the reconnect attempts are logged at WARN. Set BackoffPolicy::max_attempts if the worker should exit instead.
• Jobs in flight across an outage are redelivered. The broker requeues everything that was unacknowledged when the connection went down, so a job whose handler was still running is run again on the new connection, and the settle its first run eventually attempts fails (counted by WorkerHandle::settle_failures). That is the at-least-once contract this backend already has, but an outage is when it actually bites.

Set to None for the original behaviour: the connection is not rebuilt, consumer streams end, and Worker::run returns an error.

Implementations

impl RabbitMqOptions

pub fn connection_properties(self, properties: ConnectionProperties) -> Self

Replace the connection handshake properties.

pub fn dead_suffix(self, suffix: impl Into<String>) -> Self

Replace the dead-letter queue suffix.

pub fn declare_dead_letter_queues(self, declare: bool) -> Self

Enable or disable declaring q.dead queues.

pub fn deferred_suffix(self, suffix: impl Into<String>) -> Self

Replace the hold queue infix.

pub fn retry_granularity(self, granularity: Duration) -> Self

Replace the step retry backoffs and delayed enqueues are rounded up to.

A zero or sub-millisecond value is clamped to one millisecond when the TTL is computed, not rejected here: this is a builder, and library code does not panic on configuration.

pub fn deferred_granularity(self, granularity: Duration) -> Self

Replace the step deferral delays are rounded up to.

A zero or sub-millisecond value is clamped to one millisecond when the TTL is computed, not rejected here: this is a builder, and library code does not panic on configuration.

pub fn reconnect(self, policy: Option<Arc<dyn ReconnectPolicy>>) -> Self

Replace the reconnection policy, or pass None to disable reconnection entirely.

Takes anything that is already an Arc<dyn ReconnectPolicy>; use reconnect_with to pass a policy by value.

use std::sync::Arc;
use queuey_rabbitmq::{BackoffPolicy, RabbitMqOptions, ReconnectPolicy};

let policy: Arc<dyn ReconnectPolicy> =
    Arc::new(BackoffPolicy::default().max_attempts(Some(5)));
let bounded = RabbitMqOptions::default().reconnect(Some(policy));
assert!(bounded.reconnect.is_some());

// Or fail fast, as this backend did before reconnection existed.
let never = RabbitMqOptions::default().reconnect(None);
assert!(never.reconnect.is_none());

pub fn reconnect_with(self, policy: impl ReconnectPolicy + 'static) -> Self

Reconnect according to policy, wrapping it for you.

The common case: the backend stores policies behind an Arc because every consumer and publisher consults the same one, but a caller building options should not have to say so.

use std::time::Duration;
use queuey_rabbitmq::{Attempt, BackoffPolicy, RabbitMqOptions, ReconnectPolicy};

// The built-in policy, tuned.
let bounded = RabbitMqOptions::default()
    .reconnect_with(BackoffPolicy::default().max_attempts(Some(5)));

// Or one of your own.
#[derive(Debug)]
struct EverySecond;
impl ReconnectPolicy for EverySecond {
    fn next_delay(&self, _: Attempt<'_>) -> Option<Duration> {
        Some(Duration::from_secs(1))
    }
}
let steady = RabbitMqOptions::default().reconnect_with(EverySecond);
assert!(steady.reconnect.is_some());

Trait Implementations

impl Clone for RabbitMqOptions

fn clone(&self) -> RabbitMqOptions

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for RabbitMqOptions

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

Formats the value using the given formatter.

impl Default for RabbitMqOptions

fn default() -> Self

Returns the “default value” for a type.