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 is out of scope: when the connection is lost, consumer streams end and subsequent operations fail.

See 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.

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 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).

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,

Start consuming queue with the given prefetch. The stream ends when the backend is closed or the connection is lost.

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

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

Graceful shutdown: stop all consumers, flush, close connections.

impl Debug for RabbitMqBackend

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

Formats the value using the given formatter.