Struct RabbitMqBackend 

pub struct RabbitMqBackend { /* private fields */ }

A Backend backed by a single RabbitMQ connection.

• One Connection.
• One publishing Channel in confirm mode, shared behind a tokio::sync::Mutex. Every publish (enqueue, retry, defer, dead-letter) is mandatory and waits for the broker’s confirmation. The channel is reopened lazily if a channel exception closed it. Nothing is ever declared on it.
• One long-lived Channel for the hold queue declarations every retry, delayed publish and defer makes on demand. A declaration is the one thing the broker routinely refuses (PRECONDITION_FAILED closes the channel it ran on), so it is kept away from the publishes it would otherwise take down with it.
• One fresh Channel per consume call, so each consumer gets its own basic_qos prefetch window and a failure on one consumer cannot take down the others.
• One short-lived channel per declare call, so a rejected declaration (e.g. re-declaring an existing queue with different arguments, which RabbitMQ answers with PRECONDITION_FAILED and closes the channel) cannot poison the other channels.

Reconnection

The connection above is not one socket but a slot. When it drops, the first operation to notice dials a replacement, the others queue behind it, every queue this backend declared is re-declared on it, and the consumer streams resubscribe and carry on yielding. Nothing above the backend sees an error: a publish issued during the outage waits, and Worker::run keeps running.

What does not survive is anything already in flight. The broker requeues every unacknowledged delivery when a connection drops, so a job whose handler was mid-run is delivered again on the new connection, and when the first run finishes and tries to settle, that settle fails (the worker counts it in WorkerHandle::settle_failures). Handlers were already required to tolerate this — the contract is at-least-once — but an outage is when it stops being theoretical.

See RabbitMqOptions::reconnect to bound the attempts or turn it off, and topology for the queues this creates.

Implementations

impl RabbitMqBackend

pub async fn connect(uri: &str) -> Result<Self>

Connect to uri with RabbitMqOptions::default.

use queuey_rabbitmq::RabbitMqBackend;

let backend = RabbitMqBackend::connect("amqp://guest:guest@localhost:5672/%2f").await?;

pub async fn with_options(uri: &str, options: RabbitMqOptions) -> Result<Self>

Connect to uri with explicit options.

Only the first connection is made here, and it is not retried: a process that cannot reach its broker at startup should fail loudly rather than block its caller in a backoff loop. Every connection after this one is RabbitMqOptions::reconnect’s business.

pub fn is_connected(&self) -> bool

Whether the backend currently holds a live connection.

A false does not mean the backend is broken: with reconnection on (the default) the next operation waits for a replacement. It is here for health endpoints and dashboards that want to report the gap rather than cause one, and it never itself triggers a reconnect.

pub fn options(&self) -> &RabbitMqOptions

The options this backend was built with.

pub fn dead_queue_name(&self, queue: &str) -> String

The name of the dead-letter queue backing queue.

pub fn deferred_queue_name(&self, queue: &str, ttl_ms: u32) -> String

The name of the hold queue that ttl_ms-long waits of queue happen in.

There is one per distinct rounded delay, shared by retries and deferrals, and it is created on demand by whatever schedules the wait rather than by declare; see topology.

Trait Implementations

impl Backend for RabbitMqBackend

fn publish<'life0, 'life1, 'async_trait>( &'life0 self, envelope: &'life1 Envelope, delay: Option<Duration>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Publish envelope to its queue, or with delay into the hold queue that releases it onto its queue afterwards.

A delayed publish is held exactly like a retry: the delay is rounded up to retry_granularity, the job returns at the priority the envelope carries (0 for a fresh envelope, so it joins the back of the queue), and the same two rules as for defer apply: the queue must have been declared through this backend, and the delay must not exceed MAX_DEFERRAL_MS. An undelayed publish has neither restriction.

fn defer<'life0, 'life1, 'async_trait>( &'life0 self, envelope: &'life1 Envelope, delay: Duration, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Hold envelope for delay, then put it back on its own queue.

The queue must have been declared through this backend

Deferring onto a queue this backend instance never declared is Error::UnknownQueue, not a best-effort publish. A hold queue has to know its main queue’s durability and dead-letter it back by name, and neither can be guessed: a transient hold queue in front of a durable queue loses jobs on a restart, and a TTL expiry into a queue that does not exist is dropped by the broker in silence. Unlike a mandatory publish, nothing is returned and nothing is reported.

Producer::new and WorkerBuilder::build declare the whole queue set, so anything built through them can defer. Producer::new_undeclared deliberately does not, so a producer built that way can enqueue but cannot defer, or enqueue with a delay, until something in the process declares the queue.

The delay has a ceiling

Delays are rounded up to deferred_granularity and capped at MAX_DEFERRAL_MS (~24.8 days); a longer one is an error rather than a shorter wait.

fn consume<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 QueueConfig, ) -> Pin<Box<dyn Future<Output = Result<DeliveryStream>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Subscribe to queue and stream its deliveries.

The stream outlives the connection it started on: if the connection drops, it waits for a reconnect and resubscribes rather than ending, so Worker::run keeps going across a broker restart. It ends only when the backend is closed, or when reconnection is disabled or exhausted, which is what Error::ConsumerStopped is for.

Deliveries the worker was still holding when the connection dropped are requeued by the broker and delivered again here; settling them on the old connection fails. See the type-level docs.

fn close<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Close the connection and every channel on it, for good.

The handle is marked closing first, before a single channel goes down, so that the consumers watching for a dropped connection see a deliberate shutdown instead of an outage and end their streams rather than racing to reconnect. Nothing reopens the connection afterwards.

fn declare<'life0, 'life1, 'async_trait>( &'life0 self, queues: &'life1 [QueueConfig], ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Idempotently create all queues (plus any retry / dead-letter infrastructure).

impl Debug for RabbitMqBackend

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

Formats the value using the given formatter.